RabbitMQ消息队列-一对多模式

一对多模式，用图表示如下

一个生产者向消息队列中发送消息，多个消费者同时从消息队列中读取消息，在这个模式下，我们优先考虑的，是解决各个消费者如何读取消息的机制。

下面我们以ThinkPHP的代码来展示一下处理过程。

创建一个控制器MQSimple2，其对应的类文件MQSimple2.php中的代码如下

<?php
/**
 * Created by PhpStorm.
 * User: zhaoqinsong
 * Date: 2018/12/10
 * Time: 17:02 PM
 */
namespace app\msq\controller;
use think\App;
use think\Controller;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class MQSimple2 extends Controller
{
    var $msq_connect;
    var $msq_channel;
    public function __construct(App $app)
    {
        parent::__construct($app);
        $this->msq_connect = new AMQPStreamConnection("localhost", 5672, "phpmsq", "123456");
        $this->msq_channel = $this->msq_connect->channel();
        $this->msq_channel->queue_declare("simple", false, true, false, false);    //  生成一个持久化消息队列
    }
    public function index()
    {
        echo "rabbitmq of test";
    }
    public function mq_send() {
        //  向消息队列simple发送100个消息
        for($i=0; $i<20; $i++)
        {
            $msg = new AMQPMessage("Hello World No.$i");
            $this->msq_channel->basic_publish($msg, "", "simple");
        }
    }
    public function mq_recv1() {
        //  处理消息队列simple中的消息(时间较长)
        $callback = function($msg) {
            $body = $msg->body;
            echo "recv:$body\n";
            sleep(5);
        };
        $this->msq_channel->basic_consume("simple", "", false, true, false, false, $callback);
        while ($this->msq_channel->callbacks) {
            $this->msq_channel->wait();
        }
    }
    public function mq_recv2() {
        //  处理消息队列simple中的消息(时间较短)
        $callback = function($msg) {
            $body = $msg->body;
            echo "recv:$body\n";
            sleep(2);
        };
        $this->msq_channel->basic_consume("simple", "", false, true, false, false, $callback);
        while ($this->msq_channel->callbacks) {
            $this->msq_channel->wait();
        }
    }
    public function __destruct()
    {
        $this->msq_channel->close();
        $this->msq_connect->close();
    }
}

指定路由，设定/simple2/send指向此类的mq_send方法，/simple2/recv1指向mq_recv1方法，/simple2/recv2(处理时间5秒))指向mq_recv2方法(处理时间2秒)。

在命令行下，先创建两个服务器连接，分别执行消费者1和消费者2命令(需要先切换到ThinkPHP框架的根目录下)

# php public/index.php /simple2/recv1

# php public/index.php /simple2/recv2

# php public/index.php /simple2/send

消费者1

recv:Hello World No.0
recv:Hello World No.2
recv:Hello World No.4
recv:Hello World No.6
recv:Hello World No.8
recv:Hello World No.10
recv:Hello World No.12
recv:Hello World No.14
recv:Hello World No.16
recv:Hello World No.18

recv:Hello World No.1
recv:Hello World No.3
recv:Hello World No.5
recv:Hello World No.7
recv:Hello World No.9
recv:Hello World No.11
recv:Hello World No.13
recv:Hello World No.15
recv:Hello World No.17
recv:Hello World No.19

实际应用中，我们必须考虑各个消费者端的处理速度，因此，在这基础上，我们进行一下优化，采取basicQos以及手动应答消息完成，来实现能者多劳，按性能分配消息的机制。

修改mq_recv1和mq_recv2，修改后的相应函数代码如下

    public function mq_recv1() {
        //  处理消息队列simple中的消息(时间较长)
        $callback = function($msg) {
            $body = $msg->body;
            echo "recv:$body\n";
            sleep(5);
            $channel = $msg->delivery_info['channel'];
            $channel->basic_ack($msg->delivery_info['delivery_tag']);
        };
        $this->msq_channel->basic_qos(null, 1, null);
        $this->msq_channel->basic_consume("simple", "", false, false, false, false, $callback);
        while ($this->msq_channel->callbacks) {
            $this->msq_channel->wait();
        }
    }
    public function mq_recv2() {
        //  处理消息队列simple中的消息(时间较短)
        $callback = function($msg) {
            $body = $msg->body;
            echo "recv:$body\n";
            sleep(2);
            $channel = $msg->delivery_info['channel'];
            $channel->basic_ack($msg->delivery_info['delivery_tag']);
        };
        $this->msq_channel->basic_qos(null, 1, null);
        $this->msq_channel->basic_consume("simple", "", false, false, false, false, $callback);
        while ($this->msq_channel->callbacks) {
            $this->msq_channel->wait();
        }
    }

修改后的代码，限制每次只接受1条消息，在消息处理完成后，发送消息处理完毕的应答给消息队列，消息列表收到应答后再次发送信息给该连接，因此消息被有效地分给了各个消费者端，效率也大大增加。