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

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

 <?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;
  }
 }

 <?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()) ;
 }

 <?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()) ;
 }