消息均发

前言

---

由前文  可知，rabbitmq自带分发机制——消息会按顺序的投放到该队列下的多个消费者，例如1,3,5投放消费者C1，2,4,6投放消费者C2。

这就有个隐含的缺点：每个消息的消费时间可能不一样，极端情况下，投放给C1的每个消息消费都需要很长时间，而投放给C2的每个消息消费需要很短，就会导致C1进程

负担重,C2进程很悠闲。

所以，我们需要根据任务量来均发消息。

 

均发消息实现

---

 

1. 开启消息确认机制。

2. 为每个消费者分配且只分配一个消息，待rabbitmq收到消费者的ack后再发送后面的消息。通过这种手段，耗时很短的消息消费掉，发送ack，rabbitmq收到确认，分配下一个

消息给该消费者。不会存在很悠闲的消费者进程。

 

php代码实现

---

 

需要在消费者代码中加入 预定消息数量=1 的代码

$channel->basic_qos(null, 1, null);

整体代码如下：

sender.php

     channel();$channel->queue_declare('durable_queue', false, true, false, false);$data = implode(" ", array_slice($argv, 1));empty($data) && $data = "Hello World!";$msg = new AMQPMessage($data, array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));$channel->basic_publish($msg, '', 'durable_queue');echo " [x] Sent '$data'\n";//close the channel and connection;$channel->close();$connection->close();

receive.php

     channel();$channel->queue_declare('durable_queue', false, true, false, false);$channel->basic_qos(null, 1, null);echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";$callback = function($msg) {    echo "[x] Received ", $msg->body, "\n";    sleep(substr_count($msg->body, '.'));    echo "[x] Done\n";    //消息确认    $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);};$channel->basic_consume('durable_queue', '', false, false, false, false, $callback);while(count($channel->callbacks)) {    $channel->wait();}

 

验证

---

 

开启两个终端作为消费者C1，C2。

开启一个终端作为生产者P。

P持续生产任务量不一样的消息（用.的个数表示任务量大小）

 

看C1处理消息

 

看C2处理消息

 

 以上。