如何使用PHP和RabbitMQ实现延迟队列（方式二）？

news2024/12/23 15:45:55

前言

前几天写了一篇关于PHP和RabbitMQ如何通过插件实现延迟队列的功能。

今天写另外一篇不需要插件的方式，使用RabbitMQ的死信队列（Dead-Letter-Exchanges, DLX）和消息TTL（Time-To-Live）。

这种方法涉及到设置消息的TTL，当消息在队列中超过设定的TTL时间后，它会被发送到死信交换器。然后，你可以配置死信交换器将消息路由到另一个队列进行处理。这种方式不需要安装任何插件，但延迟时间的精度取决于消息在队列中的等待时间。

前期准备，需要安装好docker、docker-compose的运行环境。

需要安装RabbitMQ，以及通过插件实现延迟队列功能的可以浏览以下文章。

如何使用PHP和RabbitMQ实现消息队列？-CSDN博客

如何使用PHP和RabbitMQ实现延迟队列（方式一）？-CSDN博客

一、禁用RabbitMQ延迟插件功能

1、进入RabbitMQ容器，执行如下命令。

docker exec -ti rabbitmq bash

2、禁用插件，执行如下命令。

rabbitmq-plugins disable rabbitmq_delayed_message_exchange

3、检查已启用插件，执行如下命令。

rabbitmq-plugins list -e

通过下图可以看出，rabbitmq_delayed_message_exchange插件已不在启用名单中。

4、重启RabbitMQ服务。

5、至此，RabbitMQ延迟插件已禁用。

二、测试验证

1、编写生产者，代码内容如下。

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

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

// 声明延迟交换器
$delayExchange = 'delayed_exchange';
$channel->exchange_declare($delayExchange, 'direct', false, true, false);

// 声明延迟队列，并设置死信交换器和消息TTL
$delayQueue = 'delayed_queue';
$dlxExchange = 'dlx_exchange';
$delay = 5000;
$args = new \PhpAmqpLib\Wire\AMQPTable([
    'x-dead-letter-exchange' => $dlxExchange,
    'x-message-ttl' => $delay, // 消息在队列中的最大生存时间，单位毫秒
]);
$channel->queue_declare($delayQueue, false, true, false, false, false, $args);
$channel->queue_bind($delayQueue, $delayExchange, '');

// 发送消息到延迟队列
$messageBody = 'Hello Max!';
$message = new AMQPMessage($messageBody, ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]);
$channel->basic_publish($message, $delayExchange, '');

echo "Sent {$messageBody} with delay {$delay}ms\n";
$datetime = date('Y/m/d H:i:s');
echo "成功发送延迟消息 : {$messageBody} , {$datetime} \n";

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

2、编写消费者，代码内容如下。

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

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

// 声明死信交换器
$dlxExchange = 'dlx_exchange';
$channel->exchange_declare($dlxExchange, 'direct', false, true, false);

// 声明死信队列
$dlxQueue = 'dlx_queue';
$channel->queue_declare($dlxQueue, false, true, false, false);
$channel->queue_bind($dlxQueue, $dlxExchange, '');

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

// 消费者监听死信队列
$callback = function (PhpAmqpLib\Message\AMQPMessage $msg) {
    $datetime = date('Y/m/d H:i:s');
    echo "Received " . $msg->body . ",datetime=${datetime}\n";
};

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

while ($channel->is_consuming()) {
    $channel->wait();
}

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

3、启动消费端，执行如下命令。