基于Rabbitmq实现定时任务

/**
 * 死信队列  延迟队列
 * 当队列和消息同时设置过期时间 以最短时间为准
 * 消费死信队列消息
 */
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

$exchangeName = 'delay_exchange';
$queueName = 'delay_queue';
$routingKey = 'delay_key';
$deadExchangeName = 'dead_letter_exchange';
$deadQueueName = 'dead_letter_queue';
$deadRoutingKey = 'dead_letter_key';

$channel->exchange_declare($exchangeName, 'direct', false, true, false);
$channel->exchange_declare($deadExchangeName, 'direct', false, true, false);

// 创建一个一个名为delay_queue的自动过期的队列，x-message-ttl参数，如果整个队列的消息有消息都是相同的，可以设置，这里为了灵活，所以并没有设置，另外两个参数x-dead-letter-exchange代表消息过期后，消息要进入的交换机，这里配置的是delay，也就是死信交换机，x-dead-letter-routing-key是配置消息过期后，进入死信交换机的routing-key,跟发送消息的routing-key一个道理，根据这个key将消息放入不同的队列
// 死信队列参数
$arguments = new AMQPTable([
    'x-dead-letter-exchange' => $deadExchangeName,
    'x-dead-letter-routing-key' => $deadRoutingKey,
//    'x-message-ttl' => 10000
]);
$channel->queue_declare($queueName, false, true, false, false, false, $arguments);
$channel->queue_declare($deadQueueName, false, true, false, false);
$channel->queue_bind($queueName, $exchangeName, $routingKey);
$channel->queue_bind($deadQueueName, $deadExchangeName, $deadRoutingKey);

$msg = new AMQPMessage('5s后该消息将过期', ['delivery_mode' => 2, 'expiration' => 5000]);
//$channel->basic_publish($msg, 'delay', 'delay');
$channel->basic_publish($msg, $exchangeName, $routingKey);

echo " [x] Sent 'Hello World!'\n";

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

总结：基于Rabbitmq实现定时任务，就是将消息设置一个过期时间，放入一个没有读取的队列中，让消息过期后自动转入另外一个队列中，监控这个队列消息的监听处来处理定时任务具体的操作。