RabbitMq初探——消息分发

消息分发

前言

---

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

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

流程弱化如下图：

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

耗时任务

---

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

sender.php

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;

$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'
";

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

receive.php

<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLibConnectionAMQPStreamConnection;

$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', "
";

$callback = function($msg) {
    echo "[x] Received ", $msg->body, "
";
    sleep(substr_count($msg->body, '.'));
    echo "[x] Done
";
};
$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