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

前言

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

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

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

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

具体以官网为准 /d/file/titlepic/getstarted.html

介绍

config.php

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

bamq.php

 <?php /**  * created by phpstorm.  * ur: pc  * date: 2018/12/13  * time: 14:11  */  namespace myobjsummary\rabbitmq;  /** member  *  amqpchannel  *  amqpconnection  *  amqpenvelope  *  amqpexchange  *  amqpqueue  * class bamq  * @package myobjsummary\rabbitmq  */ class bamq {  /** 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   * bamq constructor.   * @throws \amqpconnectionexception   */  public function __construct()  {   $conf = require 'config.php' ;   if(!$conf)    throw new \amqpconnectionexception('config err走群众路线心得体会or!');   $this->conf  = $conf['h篮球多少钱ost'] ;   $this->exchange = $conf[乞巧节的风俗'exchange'] ;   $this->amqpconnection = new \amqpconnection($this->conf);   if (!$this->amqpconnection->connect())    throw new \amqpconnectionexception("cannot connect to the broker!\n");  }   /**   * clo link   */  public function clo()  {   $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->tname人事档案整理($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 'bamq.php'; class productmq extends bamq {  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();   $nded = true ;   foreach ($this->routes as $route) {    $nded = $ex->publish($message, $route) ;    echo "nd message:".$nded."\n";   }   if(!$nded) {    $channel->rollbacktransaction();   }   $channel->committransaction(); //提交事务   $this->clo();   die ;  } } try{  (new productmq())->run(); }catch (\exception $exception){  var_dump($exception->getmessage()) ; }

consumermq.php

 <?php //消费者 c namespace myobjsummary\rabbitmq; require 'bamq.php'; class consumermq extends bamq {  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->ttype(amqp_ex_type_direct); //direct类型   $ex->tflags(amqp_durable); //持久化   //echo "exchange status:".$ex->declare()."\n";    //创建队列   $q = $this->queue();   //var_dump($q->declare());exit();   $q->tname($this->q_name);   $q->tflags(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->clo();  } } try{  (new consumermq)->run(); }catch (\exception $exception){  var_dump($exception->getmessage()) ; }

总结

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