四、消息分发机制

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

准备工作

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

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）。