四、消息分发机制


对于计算密集型任务,需要将其分发给多个消费者进行处理。

准备工作

我们对前面测试的代码稍作改造:

task.php

<?php
/**
 * 分发任务
 */

$exchangeName = 'task';
$queueName = 'worker';
$routeKey = 'worker';
$message = empty($argv[1]) ? 'Hello World!' : $argv[1];

// 建立TCP连接
$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => '5672',
    'vhost' => '/',
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect() or die("Cannot connect to the broker!\n");

try {
    $channel = new AMQPChannel($connection);

    $exchange = new AMQPExchange($channel);
    $exchange->setName($exchangeName);
    $exchange->setType(AMQP_EX_TYPE_DIRECT);
    $exchange->declareExchange();

    echo 'Send Message: ' . $exchange->publish($message, $routeKey) . "\n";
    echo "Message Is Sent: " . $message . "\n";
} catch (AMQPConnectionException $e) {
    var_dump($e);
}

// 断开连接
$connection->disconnect();

worker.php

<?php
/**
 * 处理任务
 */

$exchangeName = 'task';
$queueName = 'worker';
$routeKey = 'worker';

// 建立TCP连接
$connection = new AMQPConnection([
    'host' => 'localhost',
    'port' => '5672',
    'vhost' => '/',
    'login' => 'guest',
    'password' => 'guest'
]);
$connection->connect() or die("Cannot connect to the broker!\n");

$channel = new AMQPChannel($connection);

$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);

echo 'Exchange Status: ' . $exchange->declareExchange() . "\n";

$queue = new AMQPQueue($channel);
$queue->setName($queueName);

echo 'Message Total: ' . $queue->declareQueue() . "\n";

echo 'Queue Bind: ' . $queue->bind($exchangeName, $routeKey) . "\n";

var_dump("Waiting for message...");

// 消费队列消息
while(TRUE) {
    $queue->consume('processMessage');
}

// 断开连接
$connection->disconnect();

function processMessage($envelope, $queue) {
    $msg = $envelope->getBody();
    var_dump("Received: " . $msg);
    sleep(substr_count($msg, '.')); // 为每一个点号模拟1秒钟操作
    $queue->ack($envelope->getDeliveryTag()); // 手动发送ACK应答
}

Round-robin dispatch 轮询分发

打开两个终端,分别运行 php worker.php,然后再开一个终端进行任务分发:

会发现任务会依次发送两个任务消费者进行处理:

第一个任务窗口

第二个任务窗口

消息确认

每个 Consumer 可能需要一段时间才能处理完收到的数据。如果在这个过程中,Consumer 出错或异常退出,而数据还没有处理完成,那么这段数据就丢失了。因为我们采用 no-ack 的方式进行确认,也就是说,每次 Consumer 接到数据后,不管是否处理完成,RabbitMQ Server 会立即把这个 Message 标记为完成,然后从 Queue 中删除。

为了保证数据不被丢失,RabbitMQ 支持消息确认机制,这种机制下不能采用 no-ack,而应该是在处理完数据后发送 ack。如果处理中途 Consumer 退出了,但是没有发送 ack,那么 RabbitMQ 就会把这个 Message 发送到下一个 Consumer,这样就保证了在 Consumer 异常退出的情况下数据也不会丢失。

这里并没有用到超时机制,RabbitMQ 仅仅通过 Consumer 的连接中断来确认该 Message 并没有被正确处理,也就是说,RabbitMQ 给 Consumer 足够长的时间来做数据处理。

之前的例子中,我们在代码中使用了 $queue->ack($envelope->getDeliveryTag());,这就是消息确认机制的应用,这种情况下,即使中断任务执行,也不会影响 RabbitMQ 中消息的处理,RabbitMQ 会将其发送给下一个 Consumer 进行处理。

如果忘记了 ack,那么后果很严重。当 Consumer 退出时,Message 会重新分发。然后 RabbitMQ 会占用越来越多的内存,由于 RabbitMQ 会长时间运行,因此这个“内存泄漏”是致命的。针对这行场景,可以通过以下命令进行 debug:

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

消息持久化

为了保证在 RabbitMQ 退出或者 crash 了数据不丢失,需要将 Queue 和 Message 持久化。

Exchange 的持久化:

$exchange->setFlags(AMQP_DURABLE);

Queue 的持久化:

$queue->setFlags(AMQP_DURABLE);

Fair dispatch 公平分发

轮询的弊病:依次分发,周而复始,在某些 Consumer 负载很重的时候,还是会分发给它。

我们可以使用 $channel->setPrefetchCount() 方法,并设置 prefetch_count = 1。这样是告诉 RabbitMQ,在同一时刻,不要发送超过 1 条消息给一个工作者(worker),直到它已经处理了上一条消息并且作出了响应。这样,RabbitMQ 就会把消息分发给下一个空闲的工作者(worker)。


Vote Vote Cancel Collect Collect Cancel

<< 上一篇: 三、PHP 中实现消息发送和接收

>> 下一篇: 五、消息订阅(Publish/Subscribe)