RabbitMQ消息队列：延迟消息

一、方案一：死信交换机 + TTL（Time-To-Live）

RabbitMQ本身并没有直接提供延迟消息的功能，但我们可以巧妙地利用 死信交换机（Dead Letter Exchange） 和 消息TTL（过期时间） 来模拟实现。

1.1 什么是死信交换机？

当一个消息在一个队列中变为“死信”时，它会被重新投递到指定的死信交换机，再由它路由到最终的队列。

消息成为死信的三种情况：

当一个队列配置了 dead-letter-exchange 属性，那么发生上述情况的消息就会被转发到该交换机。

1.2 利用死信交换机实现延迟消息的核心思想

此时，从消息发送到消费者收到，刚好经历了5秒的延迟。

1.3 方案总结

优点：

• 实现简单：利用RabbitMQ原生机制，无需安装额外插件。

• 稳定性高：基于核心功能，可靠性强。

缺点：

• 配置繁琐：需要为每个延迟任务配置死信交换机和队列。

• 时间精度不高：RabbitMQ的TTL是追溯检查的，只有当过期消息位于队首时才会被处理。如果队列前有其他消息积压，即便消息已过期，也无法及时被处理，导致延迟时间不准确。

注意：由于“队首阻塞”问题，该方案不适合对延迟时间精度要求极高的场景。

二、方案二：DelayExchange 插件（官方推荐）

鉴于方案一的局限性，RabbitMQ官方推出了 延迟消息插件（rabbitmq-delayed-message-exchange），提供了更优雅、更精准的延迟消息实现。

2.1 声明延迟交换机

我们可以声明一种新型交换机，其 delayed 属性为 true。

• 基于注解方式：

@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "delay.queue", durable = "true"),
exchange = @Exchange(name = "delay.direct", delayed = "true"),
key = "delay"
))
public void listenDelayMessage(String msg){
log.info("接收到delay.queue的延迟消息：{}", msg);
}

• 基于 @Bean 方式：

@Bean
public DirectExchange delayExchange(){
return ExchangeBuilder
.directExchange("delay.direct")
.delayed() // 关键：开启延迟特性
.durable(true)
.build();
}

2.2 发送延迟消息

发送消息时，通过设置消息头 x-delay 来指定延迟的毫秒数。

@Test
void testPublisherDelayMessage() {
String message = "hello, delayed message";
rabbitTemplate.convertAndSend("delay.direct", "delay", message, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
// 设置5秒延迟
message.getMessageProperties().setDelay(5000);
return message;
}
});
}

2.3 方案总结

优点：

• 使用简单：只需声明交换机类型，并在发送时指定延迟时间。

• 精度更高：插件内部通过Erlang定时器实现，比基于死信队列的方案更准时。

缺点：

• 依赖插件：需要额外安装。

• 性能开销：大量长延迟消息会占用插件内部数据库表和定时器资源，增加CPU开销。因此，不建议设置过长时间的延迟。

三、实战：订单支付状态同步

接下来，我们将基于 DelayExchange 插件 的方案，在“交易服务”中实现一个高可用的订单支付状态同步功能。

3.1 业务场景优化思路

30分钟的延迟消息在MQ中等待，资源消耗较大。更优方案是采用 “梯度延迟检测” 策略：
在下单后的 10秒、30秒、1分钟、2分钟、5分钟……30分钟 等多个时间点设置延迟消息。一旦在某个时间点检测到订单已支付，后续的检测任务自然取消，从而减少无效的MQ资源占用。

3.2 核心步骤

3.2.1 定义延迟消息体

为了支持“多级延迟”，我们定义一个 MultiDelayMessage 类，其中包含业务数据和一个 List<Long> 类型的延迟时间集合（单位：毫秒）。

@Data
public class MultiDelayMessage<T> {
private T data;
private List<Long> delayMillis;

// 获取并移除第一个延迟时间，实现“消费一个，取一个”的效果
public Long removeNextDelay(){
return delayMillis.remove(0);
}

public boolean hasNextDelay(){
return !delayMillis.isEmpty();
}
}

3.2.2 服务改造与配置

3.2.3 改造下单业务

在用户下单成功后，立即发送第一条延迟消息（例如10秒后）。

// 创建订单后…
// 发送延迟消息，检查支付状态
// 延迟时间数组：10秒、30秒、1分钟…
MultiDelayMessage<Long> msg = MultiDelayMessage.of(orderId, 10000L, 30000L, 60000L, …);
rabbitTemplate.convertAndSend(MqConstants.DELAY_EXCHANGE, MqConstants.DELAY_ORDER_ROUTING_KEY, msg);

3.2.4 编写支付状态查询接口

在 pay-service 中提供根据业务订单号查询支付状态的接口，并在 hm-api 模块中声明对应的 FeignClient，供交易服务远程调用。

3.2.5 核心：监听器处理逻辑

消息监听器是整个流程的大脑，其处理逻辑如下：

java

@RabbitListener(bindings = @QueueBinding(…))
public void listenOrderCheckDelayMessage(MultiDelayMessage<Long> msg) {
// 1. 获取订单ID
// 2. 本地订单状态检查
// 3. 远程查询支付状态
// 4. 支付成功，更新订单
// 5. 未支付，判断是否继续延迟检测
if (msg.hasNextDelay()) {
int delayVal = msg.removeNextDelay().intValue();
// 重新发送延迟消息，x-delay = delayVal
} else {
// 6. 超时未支付，取消订单
orderService.cancelOrder(orderId);
}
}

四、总结

本文详细介绍了RabbitMQ实现延迟消息的两种主流方案，并深入讲解了其在电商订单超时处理场景下的实战应用。

关键点回顾：