设为首页收藏本站
开启辅助访问 切换到窄版
搜索
IT技术教程资源分享平台-永久免费不加密»社区 › 编程开发教程 › PHP开发 › PHP+RabbitMQ实现消息队列的完整代码

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

  [复制链接]
查看2446 | 回复2 | 2019-8-2 10:15:22 | 显示全部楼层 |阅读模式
前言
为什么使用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
  1. <?php
  2. return [
  3. //配置
  4. 'host' => [
  5.   'host' => '127.0.0.1',
  6.   'port' => '5672',
  7.   'login' => 'guest',
  8.   'password' => 'guest',
  9.   'vhost'=>'/',
  10. ],
  11. //交换机
  12. 'exchange'=>'word',
  13. //路由
  14. 'routes' => [],
  15. ];
复制代码
BaseMQ.php
  1. <?php
  2. /**
  3. * Created by PhpStorm.
  4. * User: pc
  5. * Date: 2018/12/13
  6. * Time: 14:11
  7. */

  9. namespace MyObjSummary\rabbitMQ;

  11. /** Member
  12. *  AMQPChannel
  13. *  AMQPConnection
  14. *  AMQPEnvelope
  15. *  AMQPExchange
  16. *  AMQPQueue
  17. * Class BaseMQ
  18. * @package MyObjSummary\rabbitMQ
  19. */
  20. class BaseMQ
  21. {
  22. /** MQ Channel
  23.   * @var \AMQPChannel
  24.   */
  25. public $AMQPChannel ;

  27. /** MQ Link
  28.   * @var \AMQPConnection
  29.   */
  30. public $AMQPConnection ;

  32. /** MQ Envelope
  33.   * @var \AMQPEnvelope
  34.   */
  35. public $AMQPEnvelope ;

  37. /** MQ Exchange
  38.   * @var \AMQPExchange
  39.   */
  40. public $AMQPExchange ;

  42. /** MQ Queue
  43.   * @var \AMQPQueue
  44.   */
  45. public $AMQPQueue ;

  47. /** conf
  48.   * @var
  49.   */
  50. public $conf ;

  52. /** exchange
  53.   * @var
  54.   */
  55. public $exchange ;

  57. /** link
  58.   * BaseMQ constructor.
  59.   * @throws \AMQPConnectionException
  60.   */
  61. public function __construct()
  62. {
  63.   $conf = require 'config.php' ;
  64.   if(!$conf)
  65.    throw new \AMQPConnectionException('config error!');
  66.   $this->conf  = $conf['host'] ;
  67.   $this->exchange = $conf['exchange'] ;
  68.   $this->AMQPConnection = new \AMQPConnection($this->conf);
  69.   if (!$this->AMQPConnection->connect())
  70.    throw new \AMQPConnectionException("Cannot connect to the broker!\n");
  71. }

  73. /**
  74.   * close link
  75.   */
  76. public function close()
  77. {
  78.   $this->AMQPConnection->disconnect();
  79. }

  81. /** Channel
  82.   * @return \AMQPChannel
  83.   * @throws \AMQPConnectionException
  84.   */
  85. public function channel()
  86. {
  87.   if(!$this->AMQPChannel) {
  88.    $this->AMQPChannel = new \AMQPChannel($this->AMQPConnection);
  89.   }
  90.   return $this->AMQPChannel;
  91. }

  93. /** Exchange
  94.   * @return \AMQPExchange
  95.   * @throws \AMQPConnectionException
  96.   * @throws \AMQPExchangeException
  97.   */
  98. public function exchange()
  99. {
  100.   if(!$this->AMQPExchange) {
  101.    $this->AMQPExchange = new \AMQPExchange($this->channel());
  102.    $this->AMQPExchange->setName($this->exchange);
  103.   }
  104.   return $this->AMQPExchange ;
  105. }

  107. /** queue
  108.   * @return \AMQPQueue
  109.   * @throws \AMQPConnectionException
  110.   * @throws \AMQPQueueException
  111.   */
  112. public function queue()
  113. {
  114.   if(!$this->AMQPQueue) {
  115.    $this->AMQPQueue = new \AMQPQueue($this->channel());
  116.   }
  117.   return $this->AMQPQueue ;
  118. }

  120. /** Envelope
  121.   * @return \AMQPEnvelope
  122.   */
  123. public function envelope()
  124. {
  125.   if(!$this->AMQPEnvelope) {
  126.    $this->AMQPEnvelope = new \AMQPEnvelope();
  127.   }
  128.   return $this->AMQPEnvelope;
  129. }
  130. }
复制代码
ProductMQ.php
  1. <?php
  2. //生产者 P
  3. namespace MyObjSummary\rabbitMQ;
  4. require 'BaseMQ.php';
  5. class ProductMQ extends BaseMQ
  6. {
  7. private $routes = ['hello','word']; //路由key

  9. /**
  10.   * ProductMQ constructor.
  11.   * @throws \AMQPConnectionException
  12.   */
  13. public function __construct()
  14. {
  15.   parent::__construct();
  16. }

  18. /** 只控制发送成功 不接受消费者是否收到
  19.   * @throws \AMQPChannelException
  20.   * @throws \AMQPConnectionException
  21.   * @throws \AMQPExchangeException
  22.   */
  23. public function run()
  24. {
  25.   //频道
  26.   $channel = $this->channel();
  27.   //创建交换机对象
  28.   $ex = $this->exchange();
  29.   //消息内容
  30.   $message = 'product message '.rand(1,99999);
  31.   //开始事务
  32.   $channel->startTransaction();
  33.   $sendEd = true ;
  34.   foreach ($this->routes as $route) {
  35.    $sendEd = $ex->publish($message, $route) ;
  36.    echo "Send Message:".$sendEd."\n";
  37.   }
  38.   if(!$sendEd) {
  39.    $channel->rollbackTransaction();
  40.   }
  41.   $channel->commitTransaction(); //提交事务
  42.   $this->close();
  43.   die ;
  44. }
  45. }
  46. try{
  47. (new ProductMQ())->run();
  48. }catch (\Exception $exception){
  49. var_dump($exception->getMessage()) ;
  50. }
复制代码
ConsumerMQ.php
  1. <?php
  2. //消费者 C
  3. namespace MyObjSummary\rabbitMQ;
  4. require 'BaseMQ.php';
  5. class ConsumerMQ extends BaseMQ
  6. {
  7. private $q_name = 'hello'; //队列名
  8. private $route = 'hello'; //路由key

  10. /**
  11.   * ConsumerMQ constructor.
  12.   * @throws \AMQPConnectionException
  13.   */
  14. public function __construct()
  15. {
  16.   parent::__construct();
  17. }

  19. /** 接受消息 如果终止 重连时会有消息
  20.   * @throws \AMQPChannelException
  21.   * @throws \AMQPConnectionException
  22.   * @throws \AMQPExchangeException
  23.   * @throws \AMQPQueueException
  24.   */
  25. public function run()
  26. {

  28.   //创建交换机
  29.   $ex = $this->exchange();
  30.   $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型
  31.   $ex->setFlags(AMQP_DURABLE); //持久化
  32.   //echo "Exchange Status:".$ex->declare()."\n";

  34.   //创建队列
  35.   $q = $this->queue();
  36.   //var_dump($q->declare());exit();
  37.   $q->setName($this->q_name);
  38.   $q->setFlags(AMQP_DURABLE); //持久化
  39.   //echo "Message Total:".$q->declareQueue()."\n";

  41.   //绑定交换机与队列，并指定路由键
  42.   echo 'Queue Bind: '.$q->bind($this->exchange, $this->route)."\n";

  44.   //阻塞模式接收消息
  45.   echo "Message:\n";
  46.   while(True){
  47.    $q->consume(function ($envelope,$queue){
  48.     $msg = $envelope->getBody();
  49.     echo $msg."\n"; //处理消息
  50.     $queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答
  51.    });
  52.    //$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答
  53.   }
  54.   $this->close();
  55. }
  56. }
  57. try{
  58. (new ConsumerMQ)->run();
  59. }catch (\Exception $exception){
  60. var_dump($exception->getMessage()) ;
  61. }
复制代码


RabbitMQ, 消息队列

相关帖子

主题推荐
回复

使用道具 举报

kangly | 2019-10-23 09:22:57 | 显示全部楼层
强烈支持楼主ing……
回复

使用道具 举报

beauty | 2021-4-21 10:03:47 | 显示全部楼层

强烈支持楼主ing……
回复

使用道具 举报

返回列表 发新帖
高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

fengzi

222

主题

232

帖子

3万

积分

管理员

Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127Rank: 127

积分
38813
回复楼主 返回列表