PHP使用RabbitMQ入门2——工作队列

文章目录 收缩

上一篇文章《PHP使用RabbitMQ入门1——hello,world》通过最简单的示例跑起来了,可以对php如何使用RabbitMQ有了一个简单的了解。本篇文章主要讲解创建一个工作队列(Work Queue),它会发送一些耗时的任务给多个工作者(Worker)。
有时候我们需要处理一些耗时任务,例如裁剪图片,转换pdf,发送邮件,下载文件等,都需要较长的时间。例如用户在点击发送验证码的按钮时,完全可以将要发送的邮件内容写入到RabbitMQ的队列中,然后无需等待发送邮件,就可以直接返回,提示用户去邮箱查看。然后后台worker程序再从RabbitMQ的队列中取出邮件内容执行发送操作。这样一来,就将原本需要同步等待的操作变为了异步操作,节省了时间,提高了用户体验。
下面就以发送邮件为例来讲解RabbitMQ的工作队列。

创建生产者:send.php

创建send.php文件用于将邮件内容推送至RabbitMQ的工作队列。

<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection('x.x.x.x', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare('sendemail', false, false, false, false);

$code = mt_rand(1000,9999);
$emailAddress = implode(' ', array_slice($argv, 1));

$msg = new AMQPMessage('你的验证码是:'.$code.'=>'.$emailAddress);
$channel->basic_publish($msg, '', 'sendemail');

echo " send email to {$emailAddress} \n";

$channel->close();
$connection->close();

执行程序:

php send.php 884358@qq.com

运行结果:

在rabbitmq服务器上运行命令:

rabbitmqctl list_queues

可以看到name为sendemail的队列的message字段的值为1,代表sendemail队列有一条消息还未消费。

创建消费者:worker.php

<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$connection = new AMQPStreamConnection('x.x.x.x', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('sendemail', false, false, false, false);

echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";

$callback = function($msg) {
  echo " [√] Received ", $msg->body, "\n";
  $arr = explode('=>',$msg->body);
  echo " [√] Send email to ", $arr[1], "...\n";
  sleep(2);
  echo " [√] Done\n";
};
$channel->basic_consume('sendemail', '', false, true, false, false, $callback);

while(count($channel->callbacks)) {
    $channel->wait();
}

执行程序:

php worker.php


消费者运行后,rabbitmq会将队列中未消费的消息推送给消费者,并将消息删除。
再到rabbitmq服务器上运行命令:

rabbitmqctl list_queues


可以看到未消费的消息变为0了。

循环调度

使用工作队列的一个好处就是它能够并行的处理队列。如果堆积了很多任务,我们只需要添加更多的工作者(workers)就可以了。
分别打开两个终端,分别输入以下命令:

php worker.php

然后再打开一个终端,运行send.php发送两次任务

php send.php 1@qq.com
php send.php 2@qq.com

执行两次send.php后,可以看到两个worker.php都获取到了任务(1个worker执行一个)

默认来说,RabbitMQ会按顺序得把消息发送给每个消费者(consumer)。平均每个消费者都会收到同等数量的消息。这种发送消息得方式叫做——轮询(round-robin)。

消息确认

当处理一个耗时的任务时,如果其中某个worker在处理任务时,挂掉了,那么消息也就丢失了。例如发送验证码时,worker在发送过程中,挂掉了,验证码没有发送成功,那么用户就不会收到验证码。而我们希望的是当其中一个worker挂掉后,RabbitMQ能把这个消息转发到另一个worker,以保证验证码可以发送成功。

为了防止消息丢失,RabbitMQ提供了消息响应机制(acknowledgments)。消费者会通过一个ack(响应),告诉RabbitMQ已经收到并处理了某条消息,然后RabbitMQ就会释放并删除这条消息。
如果worker挂掉了,没有发送响应,RabbitMQ就会认为消息没有被完全处理,然后重新发送给其他worker。这样,就算worker偶尔的挂掉,也不会丢失消息。
设置basic_consume的第四个参数为false就可以打开消息确认了:

$channel->basic_consume('sendemail', '', false, false, false, false, $callback);

同时,在$callback匿名函数中增加一行响应:

$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);

完整的worker.php代码如下(把sleep改成10秒,便于我们有足够的时间来kill进程):

<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
$connection = new AMQPStreamConnection('x.x.x.x', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare('sendemail', false, false, false, false);
echo 'Waiting for messages. To exit press CTRL+C', "\n";
$callback = function($msg) {
  echo "Received ", $msg->body, "\n";
  $arr = explode('=>',$msg->body);
  echo "Send email to ", $arr[1], "...\n";
  sleep(10);
  echo "Done\n";
  $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$channel->basic_consume('sendemail', '', false, false, false, false, $callback);
while(count($channel->callbacks)) {
    $channel->wait();
}

运行上面的代码,即使使用CTRL+C杀掉了一个工作者(worker)进程,消息也不会丢失。

【注意】一个很容易犯的错误就是忘了basic_ack,后果很严重。消息在你的程序退出之后就会重新发送,如果它不能够释放没响应的消息,RabbitMQ就会占用越来越多的内存。
为了排除这种错误,你可以使用rabbitmqctl命令,输出messages_unacknowledged字段:

rabbitmqctl list_queues name messages_ready messages_unacknowledged

消息持久化

有些时候RabbitMQ服务器突然挂掉了,或者服务器突然重启后,原本保存在队列中的消息就会丢失,为了让服务器重启后消息的队列依旧可以使用,就需要将消息持久化保存到硬盘上。
持久化需要设置两点:
一是队列持久化。当RabbitMQ重启后,队列不会消失。
二是消息持久化。当RabbitMQ重启后,队列中的消息不会丢失。
必须将队列和消息都设置为持久化,才能保证消息不会丢失。

队列持久化

首先,为了不让队列消失,需要把队列声明为持久化(durable)。通过设置queue_declare的第三参数为true:

$channel->queue_declare('sendemail', false, true, false, false);

需要注意的是,queue_declare在send.php和worker.php中都需要修改才能生效。因为RabbitMq不允许使用不同的参数重新定义一个队列,它会返回一个错误。
通过以上设置,就可以确保在RabbitMq重启之后queue_declare队列不会丢失。

消息持久化

消息持久化需要设置send.php中的delivery_mode = 2

$msg = new AMQPMessage('Your verification code is:'.$code.'=>'.$emailAddress,
       array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
       );

通过以上设置,RabbitMq就会把队列以及消息持久化的保存到硬盘上了,这样在重启后,数据也不会丢失。下面测试下:
首先重启下RabbitMq,让之前设置的队列清空:

systemctl restart rabbitmq-server.service

然后执行send.php发送消息到RabbitMq:

php send.php 111@qq.com

在RabbitMq服务器执行命令查看未发送的消息数:

rabbitmqctl list_queues

可以看到sendemail队列有一个消息还未发送。现在重启rabbitmq服务器:

systemctl restart rabbitmq-server.service

重启后再次执行命令查看未发送的消息数:

rabbitmqctl list_queues

可以看到,未消费的消息依然存在。

公平调度

RabbitMq目前只是按轮询的方式,把消息依次发送给运行的worker,如果有的worker比较繁忙,rabbimq依然会把消息派送给他。下面举例说明,新建两个worker,分别为worker1.php和worker10.php,其中把worker1.php的sleep函数的参数设置未1,worker10.php设置为10,通过该方式来简单的演示worker10.php需要10秒才能处理完消息,而worker1只需要1秒。通过send.php向RabbitMQ发送消息,然后观察RabbitMQ的消息调度方式。
worker1.php:

<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$connection = new AMQPStreamConnection('x.x.x.x', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('sendemail', false, true, false, false);

echo 'Waiting for messages. To exit press CTRL+C', "\n";

$callback = function($msg) {
  echo "Received ", $msg->body, "\n";
  $arr = explode('=>',$msg->body);
  echo "Send email to ", $arr[1], "...\n";
  sleep(1);
  echo "Done\n";
  $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$channel->basic_consume('sendemail', '', false, false, false, false, $callback);

while(count($channel->callbacks)) {
    $channel->wait();
}

worker10.php:

<?php
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$connection = new AMQPStreamConnection('x.x.x.x', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('sendemail', false, true, false, false);

echo 'Waiting for messages. To exit press CTRL+C', "\n";

$callback = function($msg) {
  echo "Received ", $msg->body, "\n";
  $arr = explode('=>',$msg->body);
  echo "Send email to ", $arr[1], "...\n";
  sleep(10);
  echo "Done\n";
  $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$channel->basic_consume('sendemail', '', false, false, false, false, $callback);

while(count($channel->callbacks)) {
    $channel->wait();
}

分别打开两个终端,分别运行worker1.php和worker10.php。然后打开第三个终端,运行send.php发送消息。

从实验结果可以看出,虽然worker10.php更繁忙,worker1.php更闲,但rabbitmq依然会按顺序依次派发,并不会因为worker10.php更繁忙而少派发消息。
如果我们想根据worker的繁忙程度来调度消息,可以通过设置worker.php中channel对象的basic_qos方法的第二个参数为1,这样是告诉RabbitMQ,再同一时刻,不要发送超过1条消息给一个工作者(worker),直到它已经处理了上一条消息并且作出了响应。这样,RabbitMQ就会把消息分发给下一个空闲的工作者(worker)。

$channel->basic_qos(null, 1, null);

打开worker1.php和worker10.php,将以上代码加入到basic_consume的上方。再次进行以上实验:

可以看到,一共9条消息,给worker1派发了8条,只给worker10派发了1条。

参考:
工作队列

发表评论

电子邮件地址不会被公开。 必填项已用*标注