PHP操作RabbitMQ的类 exchangequeueroute kyebind

Posted 翩翩叶随风

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了PHP操作RabbitMQ的类 exchangequeueroute kyebind相关的知识,希望对你有一定的参考价值。

RabbitMQ是常见的消息中间件。也许是还是不够了解的缘故,感觉功能还好吧。

讲到队列,大家脑子里第一印象是下边这样的。

P生产者推送消息-->队列-->C消费者取出消息

结构很简单,但是RabbitMQ应该是为了丰富的功能吧,把“队列”拆分了。

分成了:exchange(交换机)和queue(队列)两个部分

同时说明:

生产者推送消息只推到exchange,不知道会进入哪个queue。

exchange通过一个route key与queue绑定,这时才会知道消息具体落到了哪个queue里。

而消费则获取消息,是直接从队列里取的。

大概就是以上这个意思,问题是还有两个特殊的说明:

1.消费者是无法订阅或者获取不存在的MessageQueue中信息。

2.消息被Exchange接受以后,如果没有匹配的Queue,则会被丢弃。

简单的理解就是,如果exchange和queue不绑定,生产者推送的消息到exchange会直接丢弃(丢失),同时consume也无法完成订阅。

所以,这里就有一个问题,无论是推送消息进队列,还是订阅消息消费,必须先定义好exchange和queue并通过route key绑定一起。

那么到底是推送消息的时候定义并绑定呢?还是订阅的时候定义并绑定呢?

根据那两个特殊的说明理解,无论是谁定义绑定,都有可能会出现问题。

所以,最终就是推送和消费之前,都尝试定义exchange和queue,并完成绑定。

 

推送消息,相对简单,就一个publish指定exchange和route key就完成了。

消费消息,相对的复杂一点,有两种方式:

1、推送(push)订阅方式,使用consume方法订阅队列,只要队列有消息就消费

2、拉取(poll)主动拉取,使用get方法,主动去从队列一次拉取一条消息

这两种情况,都有各自的应用场景,可以根据需要自行选择。

额外提醒一点:尽量不要使用循环的方式调用get方法消费队列,尤其是处理的消息很多的情况。

如果大量的消费队列,建议直接使用consume方法。

 

还有一个情况,当消费者取出消息时,可以不对消息队列做任何操作,也可以将取出的消息删除。

毕竟,队列里的消息,消费后是需要删除的,取到消息,发给队列一个然亏,队列就删除该消息。

这个取出消息后的删除,也分两种情况:

1、一种是取出就删除,consume和get两个方法都有一个参数AMQP_AUTOACK自动反馈

2、另一种是取出后,并不会自动返回删除,而是将取出来的消息处理之后确认没有问题了,手动反馈给消息队列

至于选择哪种反馈方式,根据需求自行选择。

 

关于RabbitMQ还有很多要说的,比如一个exchange可以绑定多个queue,多个exchange可以绑定一个queue(多对多的关系)

还可以根据exchange不同的模式,搭配不同的route key做不同的匹配。各种组合吧。

灵活应用起来功能还是很强大的,只是具体使用时需要仔细,因为一个不小心,不是丢失消息就是多出很多消息。

所知有限,在此不做特复杂的说明,下边的例子也是一个简单的完成一个简单的队列的操作。演示学习用。

更多的情况,就自行研究扩展。

 

RabbitMQ操作类(rabbitmq.class.php

  1 <?php
  2 // rabbitmq 操作类
  3 class RabbitMQ
  4 {
  5     // 配置变量
  6     public $configs = array(
  7         ‘host‘ => ‘localhost‘, 
  8         ‘port‘ => ‘5672‘, 
  9         ‘login‘ => ‘guest‘, 
 10         ‘password‘ => ‘guest‘,
 11         ‘vhost‘ => ‘/‘
 12     );
 13     public $exchange_name = ‘ex_q_def‘;// 交换机名称
 14     public $queue_name = ‘ex_q_def‘;// 队列名称
 15     public $route_key = ‘‘;// 路由key的名称
 16     public $durable = true;// 持久化,默认true
 17     public $autodelete = false;// 自动删除
 18 
 19     // 内部通用变量
 20     private $_conn = null;
 21     private $_exchange = null;
 22     private $_channel = null;
 23     private $_queue = null;
 24 
 25     // 构造函数
 26     public function __construct()
 27     {
 28         // 初始化队列
 29         $this->init();
 30     }
 31 
 32     // 配置rabbitmq
 33     public function set_configs($configs)
 34     {
 35         // 初始化配置
 36         if (!is_array($configs)) {
 37             echo ‘configs is not array.‘;
 38         }
 39         if (!($configs[‘host‘] && $configs[‘port‘] && $configs[‘login‘] && $configs[‘password‘])) {
 40             echo ‘configs is empty.‘;
 41         }
 42         if (!isset($configs[‘vhost‘])) {// 没有vhost元素,给出默认值
 43             $configs[‘vhost‘] = ‘/‘;
 44         } else {
 45             if (empty($configs[‘vhost‘])) {// 有vhost元素,但是值为空,给出默认值
 46                 $configs[‘vhost‘] = ‘/‘;
 47             }
 48         }
 49         $this->configs = $configs;
 50     }
 51 
 52     // 初始化rabbitmq
 53     private function init()
 54     {
 55         if (!$this->_conn) {
 56             $this->_conn = new AMQPConnection($this->configs);// 创建连接对象
 57             if (!$this->_conn->connect()) {
 58                 echo "Cannot connect to the broker 
 ";
 59                 exit(0);
 60             }
 61         }
 62 
 63         // 创建channel
 64         $this->_channel = new AMQPChannel($this->_conn);
 65     }
 66 
 67     // 创建队列(为了保证正常订阅,避免消息丢失,生产者和消费则都要尝试创建队列:交换机和队列通过路由绑定一起)
 68     public function create_queue($exchange_name=‘‘, $route_key=‘‘, $queue_name=‘‘)
 69     {
 70         if ($exchange_name != ‘‘) {
 71             // 队列名参数可以省略,默认与交换机同名
 72             $this->exchange_name = $exchange_name;// 更新交换机名称
 73             $this->queue_name = $exchange_name;// 更新队列名称
 74         }
 75         if ($route_key != ‘‘) $this->route_key = $route_key;// 更新路由
 76         if ($queue_name != ‘‘) $this->queue_name = $queue_name;// 独立更新队列名称
 77 
 78         // 创建exchange交换机
 79         $this->_exchange = new AMQPExchange($this->_channel);// 创建交换机
 80         $this->_exchange->setType(AMQP_EX_TYPE_DIRECT);// 设置交换机模式为direct
 81         if ($this->durable) {
 82             $this->_exchange->setFlags(AMQP_DURABLE);// 设置是否持久化
 83         }
 84         if ($this->autodelete) {
 85             $this->_exchange->setFlags(AMQP_AUTODELETE);// 设置是否自动删除
 86         }
 87         $this->_exchange->setName($this->exchange_name);// 设置交换机名称
 88         $this->_exchange->declare();
 89 
 90         // 创建queue队列
 91         $this->_queue = new AMQPQueue($this->_channel);
 92         if ($this->durable) {
 93             $this->_queue->setFlags(AMQP_DURABLE);// 设置是否持久化
 94         }
 95         if ($this->autodelete) {
 96             $this->_queue->setFlags(AMQP_AUTODELETE);// 设置是否自动删除
 97         }
 98         $this->_queue->setName($this->queue_name);// 设置队列名称
 99         $this->_queue->declare();// 完成队列的定义
100 
101         // 将queue和exchange通过route_key绑定在一起
102         $this->_queue->bind($this->exchange_name, $this->route_key);
103     }
104 
105     // 生产者,向队列交换机发送消息
106     public function send($msg, $exchange_name=‘‘, $route_key=‘‘, $queue_name=‘‘)
107     {
108         $this->create_queue($exchange_name, $route_key, $queue_name);
109         // 消息处理
110         if (is_array($msg)) {
111             $msg = json_encode($msg);// 将数组类型转换成JSON格式
112         } else {
113             $msg = trim(strval($msg));// 简单处理一下要发送的消息内容
114         }
115 
116         // 生产者推送消息进队列时,只能将消息推送到交换机exchange中
117         $this->_exchange->publish($msg, $this->route_key);
118     }
119 
120     // 消费者,从队列中获取数,消费队列(订阅)
121     public function run($fun_name, $exchange_name=‘‘, $route_key=‘‘, $queue_name=‘‘, $autoack=false)
122     {
123         if (!$fun_name) {// 没有返回函数,或者队列不存在
124             return false;
125         }
126         $this->create_queue($exchange_name, $route_key, $queue_name);
127         // 订阅消息
128         while (true) {
129             if ($autoack) {
130                 $this->_queue->consume($fun_name, AMQP_AUTOACK);// 自动应答
131             } else {
132                 $this->_queue->consume($fun_name);// 需要手动应答
133             }
134         }
135     }
136 
137     // 消费者,从队列中获取数,消费队列(主动获取)
138     public function get($exchange_name=‘‘, $route_key=‘‘, $queue_name=‘‘, $autoack=false)
139     {
140         $this->create_queue($exchange_name, $route_key, $queue_name);
141         // 主动获取消息
142         if ($autoack) {
143             $msg = $this->_queue->get(AMQP_AUTOACK);// 自动应答
144         } else {
145             $msg = $this->_queue->get();// 需要手动应答
146         }
147         return [‘msg‘=>$msg, ‘queue‘=>$this->_queue];
148     }
149 }
150 ?>

 

生产者推送(test_send.php)

 1 <?php
 2 require_once(‘rabbitmq.class.php‘);
 3 
 4 $rmq = new RabbitMQ;
 5 for ($i = 0; $i < 10; $i++) {
 6     echo ‘test_consume_‘ . $i .‘<br />‘;
 7     $rmq->send(‘test_consume_‘ . $i, ‘test_consume‘);
 8 }
 9 
10 for ($i = 0; $i < 10; $i++) {
11     echo ‘get_msg_‘.$i.‘<br />‘;
12     $rmq->send(‘get_msg_‘ . $i, ‘test_get‘);
13 }
14 
15 echo ‘send ok ! ‘ . date(‘Y-m-d H:i:s‘);
16 ?>

 

消费者consume(test_run.php)

 1 <?php
 2 require_once(‘rabbitmq.class.php‘);
 3 
 4 $rmq = new RabbitMQ;
 5 
 6 $s = $rmq->run(‘processMessage‘, ‘test_consume‘);
 7 
 8 function processMessage($envelope, $queue) {
 9     $msg = $envelope->getBody();
10     sleep(1);  //sleep1秒模拟任务处理
11     echo $msg."
"; //处理消息
12     $queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答
13 }
14 ?>

 

消费者get(test_get.php)

 1 <?php
 2 require_once(‘rabbitmq.class.php‘);
 3 
 4 $rmq = new RabbitMQ;
 5 
 6 $r = $rmq->get(‘test_get‘);
 7 
 8 echo $r[‘msg‘]->getBody();// 取到的消息
 9 $r[‘queue‘]->ack($r[‘msg‘]->getDeliveryTag());// 手动反馈,删除消费的消息
10 ?>

 

订阅consume可以起多个程序,队列会轮询平均的分到每一个订阅里。当然,前提是处理速度是一样,并且都有反馈。

如果处理速度不同,哪个快 ,哪就会分配更多的消息。如果没有反馈,默认只会推送3条消息,如果一直不给反馈,就不会再有推送了。

此时如果中断这个没有反馈的订阅,因为队列中没有删除,会再次分配到其他订阅者哪里继续推送消费。

同样的,如果队列中有消息,随时开启新的订阅,随时就会分配到消费的消息。

以上是关于PHP操作RabbitMQ的类 exchangequeueroute kyebind的主要内容,如果未能解决你的问题,请参考以下文章

php rabbitmq的开发体验

PHP使用RabbitMQ消息队列

使用xdebug从rabbitmq队列执行调试php

rabbitmq类

RabbitMQ 客户端开发向导

rabbitmq消息队列介绍