浅谈RabbitMQ的延迟队列

news2024/9/17 7:26:20

Part 01、延迟队列是什么

延迟队列代表了一种强大的消息传递机制，允许我们在将消息发送至RabbitMQ时，规定它们只能在未来某个预定的时间点被消费。这种特殊类型的消息被简称为"延迟消息"。

以RabbitMQ为例，它允许我们通过延迟队列实现这种消息的延迟传递和消费。通过将消息放入延迟队列，我们可以确保消息在特定时间之后才会被传递给消费者，从而实现了对消息传递的精确控制。这对于构建高效的异步任务调度、定时提醒和实现时间敏感性业务逻辑非常有价值。

Part 02、延迟队列的实现

延迟队列的实现原理实际上是将消息投递到一个普通队列中，不过该队列具有一项特殊属性：消息的消费被推迟了一段时间。这个延迟时间可以是灵活设定的，也可以是固定的。一旦消息进入队列，一个定时器开始计时；一旦计时器到达设定的时间，消息就会被移送到等待消费的队列中，准备被消费。在RabbitMQ中，提供了x-delayed-message插件为开发者快速实现延迟队列，主要包含以下几步：

1.安装插件： 确保安装了 RabbitMQ。然后，通过执行命令安装并启用 x-delayed-message 插件：

代码段：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

2.创建交换器： 使用 x-delayed-message 插件创建一个延迟交换器（Delayed Message Exchange）。这个交换器将用于处理延迟消息。

代码段：

rabbitmqadmin declare exchange name=delayed_exchange type=x-delayed-message arguments='{"x-delayed-type":"direct"}'

3.发送延迟消息： 当需要发送延迟消息时，将消息发送到刚创建的延迟交换器，并设置消息的延迟时间。这可以通过在消息头中添加 x-delayed-message 属性来实现。

代码段：

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.deliveryMode(2);
builder.headers(new HashMap<String, Object>(){{put("x-delay", 1000);}});
AMQP.BasicProperties properties = builder.build();
channel.basicPublish("delayed_exchange", "delayed_routing_key", properties, message.getBytes());

4.创建队列和绑定： 创建一个普通的队列，并将其绑定到延迟交换器上。这样，延迟交换器会根据消息的延迟时间将消息传递给队列。

代码段：

# 创建普通队列
rabbitmqadmin declare queue name=delayed_queue
# 将队列绑定到延迟交换器
rabbitmqadmin declare binding source=delayed_exchange destination_type=queue destination=delayed_queue routing_key=delayed_queu

5.消费消息： 启动一个消费者来从队列中获取延迟消息。一旦延迟时间过去，消息会被传递给消费者。

代码段：

channel.exchangeDeclare("delayed_exchange", "x-delayed-message", true, false, arguments);
channel.queueDeclare("delayed_queue", true, false, false, null);
channel.queueBind("delayed_queue", "delayed_exchange", "delayed_routing_key");
channel.basicConsume(queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.startConsuming()

总的来说，x-delayed-message 插件使得在 RabbitMQ 中实现延迟队列变得更加直观和方便，它允许将消息的延迟时间嵌入消息本身，无需使用TTL和死信队列来处理延迟消息。