在前面一章介绍了在PHP中如何使用RabbitMQ，至此入门的的部分就完成了，我们内心中一定还有很多疑问：如果多个消费者消费同一个队列怎么办？如果这几个消费者分任务的权重不同怎么办？怎么把同一个队列不同级别的任务分发给不同的消费者？如果消费者异常离线怎么办？不要着急，后面将慢慢解开面纱。我们将结合实际的应用场景来讲解更多的高级用法。

任务分发机制

设想如果把每个消息当做一个任务，生产者把任务发布到RabbitMQ，然后Consumer接收消息处理任务，如果我们发现一个Consumer不能完成任务处理怎么办呢，我们会增加Consumer的数量。由一个Consumer增加到两个Consumer，如图由C变为C1和C2共同来分单工作。如果C1和C2是完全一样的，那RabbitMQ会将任务平均分发到两个消费者。

如下我们新建c1.php和c2.php来订阅同一个队列在接收到消息后sleep1秒模拟任务处理的时间。

p.php代码，生产100条带编号的消息：

<?php

/*

 * 发布-订阅-P

 * create by superrd

 */

$queueName = 'superrd';

$exchangeName = 'superrd';

$routeKey = 'superrd';

$message = 'task--';

$connection = new AMQPConnection(array('host' => '10.99.121.137', 'port' => '5672', 'vhost' => '/', 'login' => 'superrd', 'password' => 'superrd'));

$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->setFlags(AMQP_DURABLE);

        $exchange->declareExchange();

        $queue = new AMQPQueue($channel);

        $queue->setName($queueName);

        $queue->setFlags(AMQP_DURABLE);

        $queue->declareQueue();

        $queue->bind($exchangeName, $routeKey);

        for($i=0 ; $i<100;$i++){

        $exchange->publish($message.$i,$routeKey);

        var_dump("[x] Sent $message $i");

        }

} catch (AMQPConnectionException $e) {

        var_dump($e);

        exit();

}

 $connection->disconnect();

c1.php和c2.php代码完全一样：

<?php

/*

 * 发布-订阅-c1c2

 * create by superrd

 */

$queueName = 'superrd';

$exchangeName = 'superrd';

$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);

$exchange->setFlags(AMQP_DURABLE);

$exchange->declareExchange();

$queue = new AMQPQueue($channel);

$queue->setName($queueName);

$queue->setFlags(AMQP_DURABLE);

$queue->declareQueue();

$queue->bind($exchangeName, $routeKey);

//阻塞模式接收消息

echo "Message:\n";

while(True){

        $queue->consume('processMessage');

//自动ACK应答

        //$queue->consume('processMessage', AMQP_AUTOACK);

}

$conn->disconnect();

/*

* 消费回调函数

* 处理消息

*/

function processMessage($envelope, $q) {

    $msg = $envelope->getBody();

    sleep(1);  //sleep1秒模拟任务处理

    echo $msg."\n"; //处理消息

    $q->ack($envelope->getDeliveryTag()); //手动发送ACK应答

}

打开两个中断窗口分别执行c1.php和c2.php脚本。确定两个脚本处于订阅状态，然后执行p.php脚本。

看到上面两幅图结果就一目了然了。因为两个脚本sleep的时间相同所以任务是完全平均分发到两个消费者的。我们修改下c2.php脚本的sleep时间为2秒，看下结果会怎么样。

可以看到c1.php脚本共收到66条消息，c2.php脚本收到34条消息，基本是按照2:1来分配。那RabbitMQ是如何来保证这样的分发机制呢，下面看RabbitMQ是如何通过ACK确认机制来实现任务分发的。

ACK消息确认机制

首先RabbitMQ支持消息确认机制来本证消息被consumer正常处理，当然也可以通过no-ack不使用确认机制。RabbitMQ默认是使用ACK确认机制的。当Consumer接收到RabbitMQ发布的消息时需要在适当的时机发送一个ACK确认的包来告知RabbitMQ，自己接收到了消息并成功处理。所以前面讲到适当的时机建议是在处理完消息任务后发送。正如我们之前的代码。

    $msg = $envelope->getBody();

    sleep(1);  //sleep1秒模拟任务处理

    echo $msg."\n"; //处理消息

    $q->ack($envelope->getDeliveryTag()); //手动发送ACK应答

那如果不发送会怎样呢？

在RabbitMQ中有一个prefetch_count的概念，这个参数的意思是允许Consumer最多同时处理几个任务。我的版本的RabbitMQ默认这个参数是3，也就是说如果某一个Consumer在收到消息后没有发送ACK确认包，RabbitMQ就会任务Consumer还在处理任务，当有3个消息都没有发送ACK确认包时，RabbitMQ就不会再发送消息给该Consumer。

我们把c2.php的sleep时间改回1秒，并且注释掉ACK确认。

    $msg = $envelope->getBody();

    sleep(1);  //sleep1秒模拟任务处理

    echo $msg."\n"; //处理消息

    //$q->ack($envelope->getDeliveryTag()); //手动发送ACK应答

发现c2脚本只收到三条消息。通过WEB管理工具也可以看到有三条消息是没有被ACK确认的。

当然任务并不会一直卡在这里，在这是RabbitMQ任务c2在处理这三个任务。如果c2忽然终止RabbitMQ会重新分发任务。如下我终止c2脚本。

三条任务被重新分发到了c1。再查看下WEB管理工具，unackd已经为0

如果Consumer数量很多或者希望每个Consumer同时只处理一个任务可以通过在Consumer中设置PrefetchCount来实现更加均匀的任务分发。

$channel = new AMQPChannel($connection);

$channel->setPrefetchCount(1);

如下我修改了c2的PrefetchCount为1。在WEB管理插件中可以看到已经有一个Consumer的PrefetchCount为1了。

RabbitMQ技术交流QQ群：327034977（添加时请备注RabbitMQ）