RabbitMq初探——消息均发

Posted 王大西

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMq初探——消息均发相关的知识,希望对你有一定的参考价值。

消息均发

前言


由前文 RabbitMq初探——消息分发 可知,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

<?php
/**
 * sender.php
 * Created by PhpStorm.
 * User: wangdaxi
 * Date: 2017/10/18
 * Time: 14:26
 */
require_once __DIR__ . \'/vendor/autoload.php\';
use PhpAmqpLib\\Connection\\AMQPStreamConnection;
use PhpAmqpLib\\Message\\AMQPMessage;

$connection = new AMQPStreamConnection(\'127.0.0.1\', 5672, \'guest\', \'guest\');
$channel = $connection->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

<?php
/**
 * receive.php
 * Created by PhpStorm.
 * User: wangdaxi
 * Date: 2017/10/18
 * Time: 14:34
 */
require_once __DIR__ . \'/vendor/autoload.php\';
use PhpAmqpLib\\Connection\\AMQPStreamConnection;

$connection = new AMQPStreamConnection(\'127.0.0.1\', 5672, \'guest\', \'guest\');
$channel = $connection->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处理消息

 

 以上。

 

以上是关于RabbitMq初探——消息均发的主要内容,如果未能解决你的问题,请参考以下文章

RABBITMQ初探——消息分发

RabbitMq初探——发布与订阅

SpringBoot:初探 RabbitMQ 消息队列

SpringBoot:初探 RabbitMQ 消息队列

RabbitMq初探——用队列实现RPC

RabbitMQ初探windows 环境下搭建rabbit 环境