RabbitMQ

RabbitMQ 延时方案

RabbitMQ 本身不支持延时消息，但可以通过插件实现。

---

延时插件

安装插件

RabbitMQ 的延时消息能力由插件提供。启用插件的命令是：

# 列出当前已启用的插件
rabbitmq-plugins list

# 启用延时消息交换器插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

# 启用后确认状态为 [E*]（E 表示已启用，* 表示进程正在运行）
rabbitmq-plugins list | grep delayed

验证要点

• 命令只用于验证系统状态，读者不需要记具体参数。
• 启用插件需要重启 RabbitMQ 节点；集群里每个节点都要启用，否则消息路由到未启用插件的节点会失败。
• 这个插件现在不是官方文档里的推荐方案。 官方文档站的 “Delayed Message Exchange” 页面在 RabbitMQ 3.13 与 4.x 两棵文档树里都已不存在，说明官方已不再把它当作一等公民来维护。工程上更稳妥的是下面第二节的 TTL + 死信交换器方案，它只用 RabbitMQ 的核心特性，不需要插件。

基本使用

启用插件后，多了一条 x-delayed-message 类型的交换器。声明时把延时时长作为参数传进去：

rabbitmqadmin declare exchange \
  name=order.delayed \
  type=x-delayed-message \
  durable=true \
  arguments='{"x-delayed-type": "direct"}'

声明之后，发布消息时用 x-delay 头指定要等多少毫秒：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

# 声明延时交换器（参数要和上面一致，否则报 inequivalent arg）
channel.exchange_declare(
    exchange="order.delayed",
    exchange_type="x-delayed-message",
    durable=True,
    arguments={"x-delayed-type": "direct"},
)

# 发布一条 30 分钟（1800000 毫秒）后才投递的消息
channel.basic_publish(
    exchange="order.delayed",
    routing_key="order.cancel",
    body=b'{"order_id": "ORD20241015001"}',
    properties=pika.BasicProperties(
        headers={"x-delay": 30 * 60 * 1000},
        delivery_mode=2,   # 持久化消息
    ),
)

connection.close()

几个必须注意的点：

1. x-delay 的单位是毫秒，不是秒。写错单位是这个插件最常见的低级错误——想等 30 分钟却写成 30，消息会在 30 毫秒后就被投递。
2. 消息必须持久化（delivery_mode=2）且交换器必须是 durable=true，否则节点重启时延时中的消息会丢——因为延时消息在插件内部是暂存在内存里的。
3. 延时时长有上限。插件依赖 Erlang 的定时器，延时过长的消息（通常在数小时到数天量级，具体取决于版本）可能失败或被拒绝。跨越很长时间的任务（比如”7 天后自动确认收货”）不适合用它。
4. 插件的延时精度不如轮询。它是”到点后尽力投递”，受节点负载影响；如果业务对到点精度有硬要求，需要配合确认机制与补偿扫描。

备选方案：TTL + 死信交换器

如果不想依赖插件，RabbitMQ 的核心特性也能拼出延时队列。原理是给消息设置过期时间（TTL），过期后由死信交换器（Dead Letter Exchange, DLX）转发到真正要消费的队列。

# 真正的业务队列
rabbitmqadmin declare queue name=order.cancel durable=true
rabbitmqadmin declare queue name=order.cancel.dlx durable=true \
  arguments='{"x-dead-letter-exchange": "", "x-dead-letter-routing-key": "order.cancel"}'

# 延时队列：消息在这里等过期，过期后被转发到 order.cancel
rabbitmqadmin declare queue name=order.delay durable=true \
  arguments='{"x-message-ttl": 1800000, "x-dead-letter-exchange": "", "x-dead-letter-routing-key": "order.cancel"}'

发布时把消息投递到 order.delay，它会在 30 分钟后自动出现在 order.cancel：

channel.basic_publish(
    exchange="",
    routing_key="order.delay",
    body=b'{"order_id": "ORD20241015001"}',
    properties=pika.BasicProperties(delivery_mode=2),
)

它的优势是只用核心特性，不装插件，且消息在队列里是持久化的（比插件的内存暂存可靠）。

代价是队列头部的过期消息不会被及时清理——RabbitMQ 只在消息到达队头时才检查是否过期，所以大量过期消息堆积在队列中间时会延迟投递。标准解法是按延时档位建多个队列（比如 order.delay.30m、order.delay.1h），每个队列用自己的 TTL，到期后统一转发到同一个业务队列。

优缺点

优点

• ✅ 原生分布式支持
• ✅ 高可靠性（持久化、ACK）
• ✅ 易于集成
• ✅ 延时时长可以在每条消息上单独指定，不需要为不同延时档位建不同的队列

缺点

• ❌ 需要安装额外插件
• ❌ 延时时间有限（受限于插件实现）
• ❌ 运维复杂度高
• ❌ 官方文档已不再收录该插件，长期维护性存疑
• ❌ 延时消息在插件内部暂存内存，节点重启或内存压力下有丢失风险

上一章节

消息队列延时方案

下一章节

Kafka