延迟任务应用场景
场景一：物联网系统经常会遇到向终端下发命令，如果命令一段时间没有应答，就需要设置成超时。
场景二：订单下单之后30分钟后，如果用户没有付钱，则系统自动取消订单。
场景三：过1分钟给新注册会员的用户，发送注册邮件等。
实现方案
- 定时任务轮询数据库，看是否有产生新任务，如果产生则消费任务
- pcntl_alarm为进程设置一个闹钟信号
- swoole的异步高精度定时器:swoole_time_tick(类似javascript的setInterval)和swoole_time_after(相当于javascript的setTimeout)
- rabbitmq延迟任务
以上四种方案,如果生产环境有使用到swoole建议使用第三种方案。此篇文章重点讲述第四种方案实现。
RabbitMQ延迟队列实现的方式有两种：
- 通过消息过期后进入死信交换器，再由交换器转发到延迟消费队列，实现延迟功能；
- 使用rabbitmq-delayed-message-exchange插件实现延迟功能；
注意： 延迟插件rabbitmq-delayed-message-exchange是在RabbitMQ 3.5.7及以上的版本才支持的，依赖Erlang/OPT 18.0及以上运行环境。
1.RabbitMQ死信机制实现延迟队列
RabbitMQ没有直接去实现延迟队列这个功能。而是需要通过消息的TTL和死信Exchange这两者的组合来实现。
消息的TTL（Time To Live）
消息的TTL就是消息的存活时间。RabbitMQ可以对队列和消息分别设置TTL。对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。如果队列设置了，消息也设置了，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息死亡的时间有可能不一样（不同的队列设置）。这里单讲单个消息的TTL，因为它才是实现延迟任务的关键。
可以通过设置消息的expiration字段或者队列x-message-ttl属性来设置时间，两者是一样的效果。下面例子是通过队列的ttl实现死信。
代码语言：javascript复制$queue = new AMQPQueue($channel);
$queue->setName($params['queueName']?:'');
$queue->setFlags(AMQP_DURABLE);
$queue->setArguments(array(
'x-dead-letter-exchange' => 'delay_exchange',
'x-dead-letter-routing-key' => 'delay_route',
'x-message-ttl' => 60000,
));
$queue->declareQueue();
当上面的消息扔到该队列中后，过了60秒，如果没有被消费，它就死了。不会被消费者消费到。这个消息后面的，没有“死掉”的消息对顶上来，被消费者消费。死信在队列中并不会被删除和释放，它会被统计到队列的消息数中去。单靠死信还不能实现延迟任务，还要靠Dead Letter Exchange。
Dead Letter Exchanges
Exchage的概念在这里就不在赘述，一个消息在满足如下条件下，会进死信路由，记住这里是路由而不是队列，一个路由可以对应很多队列。
1. 一个消息被Consumer拒收了，并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里，被其他消费者使用。
2. 上面的消息的TTL到了，消息过期了。
3. 队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上。
Dead Letter Exchange其实就是一种普通的exchange，和创建其他exchange没有两样。只是在某一个设置Dead Letter Exchange的队列中有消息过期了，会自动触发消息的转发，发送到Dead Letter Exchange中去。
示例
消费者 delay_consumer1.php:
代码语言：javascript复制<?php
//来源公众号：【码农编程进阶笔记】
//header('Content-Type:text/html;charset=utf8;');
$params = array(
'exchangeName' => 'delay_exchange',
'queueName' => 'delay_queue',
'routeKey' => 'delay_route',
);
$connectConfig = array(
'host' => 'localhost',
'port' => 5672,
'login' => 'guest',
'password' => 'guest',
'vhost' => '/'
);
//var_dump(extension_loaded('amqp'));
try {
$conn = new AMQPConnection($connectConfig);
$conn->connect();
if (!$conn->isConnected()) {
//die('Conexiune esuata');
//TODO 记录日志
echo 'rabbit-mq 连接错误:', json_encode($connectConfig);
exit();
}
$channel = new AMQPChannel($conn);
if (!$channel->isConnected()) {
// die('Connection through channel failed');
//TODO 记录日志
echo 'rabbit-mq Connection through channel failed:', json_encode($connectConfig);
exit();
}
$exchange = new AMQPExchange($channel);
$exchange->setFlags(AMQP_DURABLE);//声明一个已存在的交换器的，如果不存在将抛出异常，这个一般用在consume端
$exchange->setName($params['exchangeName']?:'');
$exchange->setType(AMQP_EX_TYPE_DIRECT); //direct类型
$exchange->declareExchange();
//$channel->startTransaction();
$queue = new AMQPQueue($channel);
$queue->setName($params['queueName']?:'');
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();
//绑定
$queue->bind($params['exchangeName'], $params['routeKey']);
} catch(Exception $e) {
echo $e->getMessage();
exit();
}
function callback(AMQPEnvelope $message) {
global $queue;
if ($message) {
$body = $message->getBody();
echo '接收时间：'.date("Y-m-d H:i:s", time()). PHP_EOL;
echo '接收内容：'.$body . PHP_EOL;
//为了防止接收端在处理消息时down掉，只有在消息处理完成后才发送ack消息
$queue->ack($message->getDeliveryTag());
} else {
echo 'no message' . PHP_EOL;
}
}
//$queue->consume('callback'); 第一种消费方式,但是会阻塞,程序一直会卡在此处
//注意：这里需要注意的是这个方法：$queue->consume，queue对象有两个方法可用于取消息：consume和get。前者是阻塞的，无消息时会被挂起，适合循环中使用；后者则是非阻塞的，取消息时有则取，无则返回false。
//就是说用了consume之后，会同步阻塞，该程序常驻内存，不能用nginx，apache调用。
$action = '2';
if($action == '1'){
$queue->consume('callback'); //第一种消费方式,但是会阻塞,程序一直会卡在此处
}else{
//第二种消费方式,非阻塞
$start = time();
while(true)
{
$message = $queue->get();
if(!empty($message))
{
echo '接收时间：'.date("Y-m-d H:i:s", time()). PHP_EOL;
echo '接收内容：'.$message->getBody().PHP_EOL;
$queue->ack($message->getDeliveryTag()); //应答，代表该消息已经消费
$end = time();
echo '运行时间：'.($end - $start).'秒'.PHP_EOL;
//exit();
}
else
{
//echo 'message not found' . PHP_EOL;
}
}
}
生产者delay_publisher1.php:
代码语言：javascript复制<?php
//来源公众号：【码农编程进阶笔记】
//header('Content-Type:text/html;charset=utf-8;');
$params = array(
'exchangeName' => 'test_cache_exchange',
'queueName' => 'test_cache_queue',
'routeKey' => 'test_cache_route',
);
$connectConfig = array(
'host' => 'localhost',
'port' => 5672,
'login' => 'guest',
'password' => 'guest',
'vhost' => '/'
);
//var_dump(extension_loaded('amqp')); 判断是否加载amqp扩展
//exit();
for($i=5;$i>0;$i--){
try {
$conn = new AMQPConnection($connectConfig);
$conn->connect();
if (!$conn->isConnected()) {
//die('Conexiune esuata');
//TODO 记录日志
echo 'rabbit-mq 连接错误:', json_encode($connectConfig);
exit();
}
$channel = new AMQPChannel($conn);
if (!$channel->isConnected()) {
// die('Connection through channel failed');
//TODO 记录日志
echo 'rabbit-mq Connection through channel failed:', json_encode($connectConfig);
exit();
}
$exchange = new AMQPExchange($channel);
$exchange->setFlags(AMQP_DURABLE);//持久化
$exchange->setName($params['exchangeName']);
$exchange->setType(AMQP_EX_TYPE_DIRECT); //direct类型
$exchange->declareExchange();
//$channel->startTransaction();
//RabbitMQ不容许声明2个相同名称、配置不同的Queue队列
$queue = new AMQPQueue($channel);
$queue->setName($params['queueName'].$i);
$queue->setFlags(AMQP_DURABLE);
$queue->setArguments(array(
'x-dead-letter-exchange' => 'delay_exchange', //// 死信交换机
'x-dead-letter-routing-key' => 'delay_route', // 死信路由
'x-message-ttl' => (10000*$i), // 当上面的消息扔到该队列中后，过了60秒，如果没有被消费，它就死了
// 在RMQ中想要使用优先级特性需要的版本为3.5 。
//'x-max-priority'=>0,//将队列声明为优先级队列，即在创建队列的时候添加参数 x-max-priority 以指定最大的优先级，值为0-255（整数）。
));
$queue->declareQueue();
//绑定队列和交换机
$queue->bind($params['exchangeName'], $params['routeKey'].$i);
//$channel->commitTransaction();
} catch(Exception $e) {
}
// 当mandatory标志位设置为true时，如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue，那么会调用basic.return方法将消息返还给生产者；当mandatory设为false时，出现上述情形broker会直接将消息扔掉。
//delivery_mode=2指明message为持久的
//生成消息
echo '发送时间：'.date("Y-m-d H:i:s", time()).PHP_EOL;
echo 'i='.$i.'，延迟'.($i*10).'秒'.PHP_EOL;
$message = json_encode(['order_id'=>time(),'i'=>$i]);
$exchange->publish($message, $params['routeKey'].$i, AMQP_MANDATORY, array('delivery_mode'=>2));
$conn->disconnect();
sleep(2);
}
使用方法：先运行delay_consumer1.php，再运行delay_publisher1.php
运行效果：
最关键的点就是在生产者那里
代码语言：javascript复制 $queue->setArguments(array(
'x-dead-letter-exchange' => 'delay_exchange', // 死信交换机
'x-dead-letter-routing-key' => 'delay_route', // 死信路由
'x-message-ttl' => 10000, // 当上面的消息扔到该队列中后，过了60秒，如果没有被消费，它就死了
));
详细过程:
- 首先由正常队列(test_cache_queue)和正常exchange（test_cache_exchange），两者相绑定。
- 该正常队列设置了死信路由(delay_exchange)和死信路由key以及TTL,生产者生产消息到正常队列和正常路由上.
- 当正常队列设置TTL时间一到，那延迟消息就会自动发布到死信路由
- 消费者通过死信路由(delay_exchange)和死信队列(delay_queue)来消费
注意:
代码语言：javascript复制使用死信队列实现延时消息的缺点：
1） 如果统一用队列来设置消息的TTL，当梯度非常多的情况下，比如1分钟，2分钟，5分钟，10分钟，20分钟，30分钟……需要创建很多交换机和队列来路由消息。
2） 如果单独设置消息的TTL，则可能会造成队列中的消息阻塞——前一条消息没有出队（没有被消费），后面的消息无法投递。
3） 可能存在一定的时间误差。
4） ttl设置之后，下次修改时间，会报错，这时候，需要先删除该队列，重启项目。否则会报错。
5） 消费者中，抛异常了没处理，会一直重复消费
