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

php  /  管理员 发布于 4年前   400

前言

为什么使用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

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

BaseMQ.php

 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

 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

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

总结

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

您可能感兴趣的文章:

• php实现通过stomp协议连接ActiveMQ操作示例
• php ActiveMQ的安装与使用方法图文教程
• PHP使用ActiveMQ实现消息队列的方法详解
• PHP使用ActiveMQ实例
• PHP Beanstalkd消息队列的安装与使用方法实例详解
• PHP高级编程之消息队列原理与实现方法详解
• php+redis实现消息队列功能示例
• 使用PHP访问RabbitMQ消息队列的方法示例
• php实现websocket实时消息推送
• php 使用ActiveMQ发送消息，与处理消息操作示例