欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

PHP+RabbitMQ实现消息队列的完整代码

程序员文章站 2023-12-16 10:50:58
前言 为什么使用rabbitmq而不是activemq或者rocketmq? 首先,从业务上来讲,我并不要求消息的100%接受率,并且,我需要结合php开发,rabbi...

前言

为什么使用rabbitmq而不是activemq或者rocketmq?

首先,从业务上来讲,我并不要求消息的100%接受率,并且,我需要结合php开发,rabbitmq相较rocketmq,延迟较低(微妙级)。至于activemq,貌似问题较多。rabbitmq对各种语言的支持较好,所以选择rabbitmq。

先安装php对应的rabbitmq,这里用的是 php_amqp 不同的扩展实现方式会有细微的差异.

php扩展地址: http://pecl.php.net/package/amqp

具体以官网为准  http://www.rabbitmq.com/getstarted.html

介绍

  • config.php 配置信息
  • basemq.php mq基类
  • productmq.php 生产者类
  • consumermq.php 消费者类
  • consumer2mq.php 消费者2(可有多个)

config.php

 <?php
 return [
  //配置
  'host' => [
   'host' => '127.0.0.1',
   'port' => '5672',
   'login' => 'guest',
   'password' => 'guest',
   'vhost'=>'/',
  ],
  //交换机
  'exchange'=>'word',
  //路由
  'routes' => [],
 ];

basemq.php

 <?php
 /**
  * created by phpstorm.
  * user: pc
  * date: 2018/12/13
  * time: 14:11
  */
 
 namespace myobjsummary\rabbitmq;
 
 /** member
  *  amqpchannel
  *  amqpconnection
  *  amqpenvelope
  *  amqpexchange
  *  amqpqueue
  * class basemq
  * @package myobjsummary\rabbitmq
  */
 class basemq
 {
  /** mq channel
   * @var \amqpchannel
   */
  public $amqpchannel ;
 
  /** mq link
   * @var \amqpconnection
   */
  public $amqpconnection ;
 
  /** mq envelope
   * @var \amqpenvelope
   */
  public $amqpenvelope ;
 
  /** mq exchange
   * @var \amqpexchange
   */
  public $amqpexchange ;
 
  /** mq queue
   * @var \amqpqueue
   */
  public $amqpqueue ;
 
  /** conf
   * @var
   */
  public $conf ;
 
  /** exchange
   * @var
   */
  public $exchange ;
 
  /** link
   * basemq constructor.
   * @throws \amqpconnectionexception
   */
  public function __construct()
  {
   $conf = require 'config.php' ;
   if(!$conf)
    throw new \amqpconnectionexception('config error!');
   $this->conf  = $conf['host'] ;
   $this->exchange = $conf['exchange'] ;
   $this->amqpconnection = new \amqpconnection($this->conf);
   if (!$this->amqpconnection->connect())
    throw new \amqpconnectionexception("cannot connect to the broker!\n");
  }
 
  /**
   * close link
   */
  public function close()
  {
   $this->amqpconnection->disconnect();
  }
 
  /** channel
   * @return \amqpchannel
   * @throws \amqpconnectionexception
   */
  public function channel()
  {
   if(!$this->amqpchannel) {
    $this->amqpchannel = new \amqpchannel($this->amqpconnection);
   }
   return $this->amqpchannel;
  }
 
  /** exchange
   * @return \amqpexchange
   * @throws \amqpconnectionexception
   * @throws \amqpexchangeexception
   */
  public function exchange()
  {
   if(!$this->amqpexchange) {
    $this->amqpexchange = new \amqpexchange($this->channel());
    $this->amqpexchange->setname($this->exchange);
   }
   return $this->amqpexchange ;
  }
 
  /** queue
   * @return \amqpqueue
   * @throws \amqpconnectionexception
   * @throws \amqpqueueexception
   */
  public function queue()
  {
   if(!$this->amqpqueue) {
    $this->amqpqueue = new \amqpqueue($this->channel());
   }
   return $this->amqpqueue ;
  }
 
  /** envelope
   * @return \amqpenvelope
   */
  public function envelope()
  {
   if(!$this->amqpenvelope) {
    $this->amqpenvelope = new \amqpenvelope();
   }
   return $this->amqpenvelope;
  }
 }

productmq.php

 <?php
 //生产者 p
 namespace myobjsummary\rabbitmq;
 require 'basemq.php';
 class productmq extends basemq
 {
  private $routes = ['hello','word']; //路由key
 
  /**
   * productmq constructor.
   * @throws \amqpconnectionexception
   */
  public function __construct()
  {
   parent::__construct();
  }
 
  /** 只控制发送成功 不接受消费者是否收到
   * @throws \amqpchannelexception
   * @throws \amqpconnectionexception
   * @throws \amqpexchangeexception
   */
  public function run()
  {
   //频道
   $channel = $this->channel();
   //创建交换机对象
   $ex = $this->exchange();
   //消息内容
   $message = 'product message '.rand(1,99999);
   //开始事务
   $channel->starttransaction();
   $sended = true ;
   foreach ($this->routes as $route) {
    $sended = $ex->publish($message, $route) ;
    echo "send message:".$sended."\n";
   }
   if(!$sended) {
    $channel->rollbacktransaction();
   }
   $channel->committransaction(); //提交事务
   $this->close();
   die ;
  }
 }
 try{
  (new productmq())->run();
 }catch (\exception $exception){
  var_dump($exception->getmessage()) ;
 }

consumermq.php

 <?php
 //消费者 c
 namespace myobjsummary\rabbitmq;
 require 'basemq.php';
 class consumermq extends basemq
 {
  private $q_name = 'hello'; //队列名
  private $route = 'hello'; //路由key
 
  /**
   * consumermq constructor.
   * @throws \amqpconnectionexception
   */
  public function __construct()
  {
   parent::__construct();
  }
 
  /** 接受消息 如果终止 重连时会有消息
   * @throws \amqpchannelexception
   * @throws \amqpconnectionexception
   * @throws \amqpexchangeexception
   * @throws \amqpqueueexception
   */
  public function run()
  {
 
   //创建交换机
   $ex = $this->exchange();
   $ex->settype(amqp_ex_type_direct); //direct类型
   $ex->setflags(amqp_durable); //持久化
   //echo "exchange status:".$ex->declare()."\n";
 
   //创建队列
   $q = $this->queue();
   //var_dump($q->declare());exit();
   $q->setname($this->q_name);
   $q->setflags(amqp_durable); //持久化
   //echo "message total:".$q->declarequeue()."\n";
 
   //绑定交换机与队列,并指定路由键
   echo 'queue bind: '.$q->bind($this->exchange, $this->route)."\n";
 
   //阻塞模式接收消息
   echo "message:\n";
   while(true){
    $q->consume(function ($envelope,$queue){
     $msg = $envelope->getbody();
     echo $msg."\n"; //处理消息
     $queue->ack($envelope->getdeliverytag()); //手动发送ack应答
    });
    //$q->consume('processmessage', amqp_autoack); //自动ack应答
   }
   $this->close();
  }
 }
 try{
  (new consumermq)->run();
 }catch (\exception $exception){
  var_dump($exception->getmessage()) ;
 }

总结

以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,谢谢大家对的支持。

上一篇:

下一篇: