在上一篇文章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
网友评论