RABBITMQ初探——消息分发

Posted 2020-10-11 王大西

消息分发

前言

---

 

我们在用到消息队列的场景，一般是处理逻辑复杂，耗时，所以将同步改为异步处理，接入队列，下游处理耗时任务。

队列消息数量很大，且下游worker进程（消费者）处理耗时长，所以就有了任务的积压。rabbitmq提供了任务分发的机制。

流程弱化如下图：

可以接入多个消费者，rabbitmq会将消息均匀的分发给每一个消费者。

耗时任务

---

 

我们可以在consumer端用sleep()函数来模拟耗时任务，通过判断消息的点的个数，来进行相应的sleep几秒。

sender.php

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(\'hello\', false, false, false, false);

$data = implode(" ", array_slice($argv, 1));
empty($data) && $data = "Hello World!";

$msg = new AMQPMessage($data);

$channel->basic_publish($msg, \'\', \'hello\');

echo " [x] Sent \'$data\'\\n";

//close the channel and connection;
$channel->close();
$connection->close();

 

receive.php

<?php
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(\'hello\', false, false, false, false);
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";
};
$channel->basic_consume(\'hello\', \'\', false, true, false, false, $callback);

while(count($channel->callbacks)) {
    $channel->wait();
}

 

 

验证

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

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

P1生产消息：

C1消费消息：

 C2消费消息：

 

 以上，是rabbitmq的 Round-robin dispatching