RabbitMQ初探(四):消息分发机制

在上一篇文章RabbitMQ初探(三)中我们一个终端执行rabbit_consumer.php(消费脚本)一个终端执行rabbit_publisher(发布脚本),那如果有多个消费终端同时执行呢？

继续上一篇的代码

rabbit_consumer.php

<?php 
//配置信息 
$conn_args = array( 
    'host' => '127.0.0.1',  
    'port' => '5672',  
    'login' => 'guest',  
    'password' => 'guest', 
    'vhost'=>'/' 
);   
$ex_name = 'e_linvo'; //交换机名 
$queue_name = 'q_linvo'; //队列名 
$k_route = 'key_1'; //路由key 
 
//创建连接和channel 
$conn = new AMQPConnection($conn_args);   
if (!$conn->connect()) {   
    die("Cannot connect to the broker!" . PHP_EOL);   
}   
$channel = new AMQPChannel($conn);   
 
//创建交换机    
$ex = new AMQPExchange($channel);   
$ex->setName($ex_name); 
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型   如果 routing key 匹配，那么 Message 就会被传递到相应的 queue
$ex->setFlags(AMQP_DURABLE); //持久化 
echo "Exchange Status:" . $ex->declareExchange() . PHP_EOL;   
   
//创建队列    
$q = new AMQPQueue($channel); 
$q->setName($queue_name);   
$q->setFlags(AMQP_DURABLE); //持久化  
echo "Message Total:".$q->declareQueue() . PHP_EOL;   
 
//绑定交换机与队列，并指定路由键 
echo 'Queue Bind: '.$q->bind($ex_name, $k_route) . PHP_EOL; 
 
//阻塞模式接收消息 
echo "Message:" . PHP_EOL;   
while(True){ 
    $q->consume('processMessage');   
    //$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答  
} 
$conn->disconnect();   
 
/**
 * 消费回调函数
 * 处理消息
 */ 
function processMessage($envelope, $queue) { 
    $msg = $envelope->getBody(); 
    echo $msg . PHP_EOL; //处理消息 
    $queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答 
}

rabbit_publisher 循环次数稍作修改 改为10次

<?php
//connect的配置信息 
$conn_args = [
    'host' => '127.0.0.1',  
    'port' => '5672',  
    'login' => 'guest',  
    'password' => 'guest', 
    'vhost'=>'/' 
];   
$ex_name = 'e_linvo'; //交换机名 
$k_route = 'key_1'; //路由key 
 
//创建连接和channel 
$conn = new AMQPConnection($conn_args);   
if (!$conn->connect()) {   
    die("Cannot connect to the broker!\n");   
}   
$channel = new AMQPChannel($conn);   
 
//创建交换机对象    
$ex = new AMQPExchange($channel);   
$ex->setName($ex_name);   
date_default_timezone_set("Asia/Shanghai");
//发送消息  
for($i=0; $i<10; ++$i){ 
    sleep(1);//休眠1秒
    //消息内容 
    $message = "TEST MESSAGE!" . date("h:i:sa");   
    echo "Send Message:" . $ex->publish($message, $k_route) . PHP_EOL;  
} 
 
$conn->disconnect();

先开启2个终端执行php rabbit_consumer.php然后开启一个终端执行php rabbit_publisher.php执行结果如下:

发布端

Send Message:1
Send Message:1
Send Message:1
Send Message:1
Send Message:1
Send Message:1
Send Message:1
Send Message:1
Send Message:1
Send Message:1

消费端1

Exchange Status:1
Message Total:0
Queue Bind: 1
Message:
TEST MESSAGE!11:55:07
TEST MESSAGE!11:55:07
TEST MESSAGE!11:55:07
TEST MESSAGE!11:55:07
TEST MESSAGE!11:55:07

消费端2

Exchange Status:1
Message Total:0
Queue Bind: 1
Message:
TEST MESSAGE!11:55:07
TEST MESSAGE!11:55:07
TEST MESSAGE!11:55:07
TEST MESSAGE!11:55:07
TEST MESSAGE!11:55:07

我们看到消息是平均分配到2个队列并执行的，也就是Fair dispatch公平分发。我们很容易想到他的弊病:依次分发，周而复始，在某些 Consumer 负载很重的时候，还是会分发给它,那我们有没有什么办法可以让某一个Consumer负载重的时候不接受分发呢。

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

我们修改rabbit_consumer.php文件 在$channel = new AMQPChannel($conn); 下面追加一行代码

$channel->setPrefetchCount(2);   //每个消费者队列最多同时有2条数据待处理，

消费端1

Exchange Status:1
Message Total:0
Queue Bind: 1
Message:
TEST MESSAGE!11:57:46
TEST MESSAGE!11:57:46
TEST MESSAGE!11:57:46
TEST MESSAGE!11:57:46
TEST MESSAGE!11:57:46
TEST MESSAGE!11:57:46

消费端2

Exchange Status:1
Message Total:0
Queue Bind: 1
Message:
TEST MESSAGE!11:57:46
TEST MESSAGE!11:57:46
TEST MESSAGE!11:57:46
TEST MESSAGE!11:57:46