RabbitMQ 延时消息队列实现文档

功能：下单后超时未支付，自动取消订单
模块：mall-order
技术方案：Spring AMQP + RabbitMQ 延时消息插件（rabbitmq-delayed-message-exchange）+ 多级延迟投递

---

目录

• 一、需求背景与方案选型
• 二、整体架构
• 三、环境准备
• 四、引入依赖
• 五、连接配置
• 六、消息转换器（JSON 序列化）
• 七、常量定义（交换机 / 队列 / 路由键）
• 八、消息体结构 MultiDelayMessage
• 九、生产者：下单发送延时消息
• 十、消费者：检查订单 / 重投 / 取消
• 十一、完整执行时序
• 十二、关键设计点
• 十三、不足与生产改进建议
• 十四、相关文件清单

---

一、需求背景与方案选型

业务要求：用户下单后若在规定时间内未支付，系统应自动关闭该订单、释放库存。

常见方案对比

方案	说明	缺点
定时任务轮询	每隔一段时间扫描整个订单表，找出超时订单	数据量大时性能差、定时存在间隔，关闭不及时
死信队列 TTL + DLX	消息设 TTL，过期后经死信交换机转发	同一队列 TTL 不一致时会有队头阻塞问题
延时消息插件（本项目）	消息携带 x-delay，由插件按每条消息独立计时	需安装插件；极高堆积场景不适合

本项目最终采用 延时消息插件，并在其基础上设计了多级延迟：消息到期后若订单仍未支付，就用下一档延迟重新投递，直到所有延迟梯度耗尽再取消订单。

为什么要”多级延迟”

• 用户在中途完成支付时，最近的一档检查就能发现状态变化，消息链随即终止，不必一直等到最终超时；
• 用一组递增的延迟梯度（10s → 10s → 10s → 15s → 15s → 30s → 30s）兼顾了”及时发现”与”减少无效检查次数”。

---

二、整体架构

flowchart LR
    A[下单 OrderServiceImpl.add] -->|convertAndSend<br/>x-delay=10s| X
    subgraph MQ[RabbitMQ Broker]
        X[延时交换机<br/>trade.delay.topic] -->|routing key<br/>order.query| Q[延时队列<br/>trade.order.delay.queue]
    end
    Q -->|到期投递| L[OrderStatusListener]
    L -->|仍未支付且有剩余梯度| X
    L -->|延迟耗尽仍未支付| C[cancelOrder<br/>取消订单]
    L -->|已支付 / 订单不存在| D[丢弃消息]

消息流转三要素

• 延时交换机 trade.delay.topic：声明为 delayed = "true"，由插件实现，拦截带 x-delay 头的消息并暂存计时；
• 队列 trade.order.delay.queue：到期消息最终投递到这里；
• 路由键 order.query：交换机据此把消息路由到队列。

---

三、环境准备

延时交换机依赖 RabbitMQ 延时插件，需先在 RabbitMQ 服务器安装并启用：

# 1. 下载 rabbitmq_delayed_message_exchange 插件（.ez 文件）放入 plugins 目录
# 2. 启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
# 3. 重启 RabbitMQ
rabbitmq-server restart

启用后即可声明 x-delayed-message 类型的交换机（在 Spring 中通过 @Exchange(delayed = "true") 实现）。

---

四、引入依赖

mall-order/pom.xml

<!-- RabbitMQ：spring-boot-starter-amqp，提供 RabbitTemplate、@RabbitListener 等 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

---

五、连接配置

mall-order/src/main/resources/application.yml

spring:
  rabbitmq:
    host: 127.0.0.1   # RabbitMQ 服务器地址
    port: 5672        # AMQP 通信端口（管理界面端口为 15672）
    username: guest
    password: guest
    virtual-host: /

当前为教学配置，未开启发布确认（publisher confirm）、未配置消费者并发与手动 ack：
发送属于”发完即忘”，消费端默认单线程、自动确认。

---

六、消息转换器（JSON 序列化）

mall-order/src/main/java/com/situ/mall/config/RabbitMQConfig.java

package com.situ.mall.config;

import org.springframework.amqp.support.converter.DefaultClassMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    @Bean
    public MessageConverter jsonToMessageConverter() {
        DefaultClassMapper defaultClassMapper = new DefaultClassMapper();
        // 指定反序列化时信任的包，防止恶意反序列化攻击
        defaultClassMapper.setTrustedPackages("com.situ.mall.util");

        Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
        converter.setClassMapper(defaultClassMapper);
        return converter;
    }
}

作用

• 生产者发送 MultiDelayMessage 对象时，由该转换器序列化为 JSON；
• 消费者收到后再反序列化回 MultiDelayMessage 对象；
• DefaultClassMapper 会把对象的真实类型写入消息头，因此消费者方法可以直接用具体类型 MultiDelayMessage<Long> 接收。

---

七、常量定义（交换机 / 队列 / 路由键）

mall-order/src/main/java/com/situ/mall/constant/MqConstant.java

package com.situ.mall.constant;

public interface MqConstant {
    String DELAY_EXCHANGE          = "trade.delay.topic";        // 延时交换机名
    String DELAY_ORDER_QUEUE       = "trade.order.delay.queue";  // 队列名
    String DELAY_ORDER_ROUTING_KEY = "order.query";              // 路由键
}

---

八、消息体结构 MultiDelayMessage

mall-order/src/main/java/com/situ/mall/util/MultiDelayMessage.java

一条携带订单号与多级延迟时间表的消息。

package com.situ.mall.util;

import lombok.Data;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

@Data
public class MultiDelayMessage<T> implements Serializable {

    /** 消息体：本项目中存订单号 orderNo */
    private T data;

    /** 各级延迟时间集合（单位：毫秒） */
    private List<Long> delayMillis;

    public MultiDelayMessage() {}

    public MultiDelayMessage(T data, List<Long> delayMillis) {
        this.data = data;
        this.delayMillis = delayMillis;
    }

    public MultiDelayMessage(T data, Long... delayMillis) {
        this.data = data;
        List<Long> list = new ArrayList<>();
        for (Long d : delayMillis) {
            list.add(d);
        }
        this.delayMillis = list;
    }

    /** 取出并移除第一个延迟时间（用掉一档） */
    public Long removeNextDelay() {
        return delayMillis.remove(0);
    }

    /** 是否还有剩余延迟档 */
    public boolean hasNextDelay() {
        return !delayMillis.isEmpty();
    }
}

• data 使用泛型，本项目实际为 MultiDelayMessage<Long>，装载雪花算法生成的订单号；
• delayMillis 保存延迟梯度，每消费一次通过 removeNextDelay() 用掉一档。

---

九、生产者：下单发送延时消息

mall-order/src/main/java/com/situ/mall/service/impl/OrderServiceImpl.java（核心片段）

@Autowired
private RabbitTemplate rabbitTemplate;

@Transactional(propagation = Propagation.REQUIRED)
@Override
public void add(Order order) {
    // ... 此处省略：获取购物车勾选商品、计算总价、
    //              orderMapper.insert(order)、orderItemMapper.insert(orderItem)、删除购物车 ...

    // ===== 发送延时消息 =====
    // 构造消息：data=订单号，延迟梯度 = 10s,10s,10s,15s,15s,30s,30s
    MultiDelayMessage<Long> msg = new MultiDelayMessage<>(
            order.getOrderNo(),
            10000L, 10000L, 10000L, 15000L, 15000L, 30000L, 30000L);

    // 用掉第一档延迟（10 秒）
    Long delay = msg.removeNextDelay();

    // 发送到延时交换机，并通过 MessagePostProcessor 设置消息头 x-delay
    rabbitTemplate.convertAndSend(
            MqConstant.DELAY_EXCHANGE,
            MqConstant.DELAY_ORDER_ROUTING_KEY,
            msg,
            new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties()
                            .getHeaders()
                            .put("x-delay", delay);
                    return message;
                }
            });
}

要点

1. 消息头 x-delay 即”这一跳需要延迟的毫秒数”，由延时插件识别；
2. 第一条消息延迟 10 秒，消息体内仍保留剩余梯度 [10000, 10000, 15000, 15000, 30000, 30000]，供后续重投使用；
3. 交换机必须为延时交换机（由消费者侧 @Exchange(delayed = "true") 声明）。

---

十、消费者：检查订单 / 重投 / 取消

mall-order/src/main/java/com/situ/mall/listener/OrderStatusListener.java

package com.situ.mall.listener;

import com.situ.mall.constant.MqConstant;
import com.situ.mall.enums.OrderStatus;
import com.situ.mall.pojo.entity.Order;
import com.situ.mall.service.IOrderService;
import com.situ.mall.util.MultiDelayMessage;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class OrderStatusListener {

    @Autowired
    private IOrderService orderService;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // @QueueBinding：应用启动时自动声明 队列、延时交换机及绑定关系
    @RabbitListener(bindings = @QueueBinding(
            value    = @Queue(MqConstant.DELAY_ORDER_QUEUE),               // 队列
            exchange = @Exchange(name = MqConstant.DELAY_EXCHANGE,
                                 delayed = "true"),                        // 延时交换机
            key      = MqConstant.DELAY_ORDER_ROUTING_KEY                 // 绑定路由键
    ))
    public void listenDelayMessage(MultiDelayMessage<Long> msg) {
        // 1. 取出订单号，查询订单当前状态
        Long orderNo = msg.getData();
        Order order = orderService.getById(orderNo);

        // 2. 订单不存在，或已不是"未支付"状态（已支付/已取消）→ 直接结束，丢弃消息
        if (order == null || order.getStatus() != OrderStatus.UNPAID.getCode()) {
            return;
        }

        // 3. 订单仍未支付，判断是否还有剩余延迟档
        if (msg.hasNextDelay()) {
            List<Long> delayMillis = msg.getDelayMillis();
            Long delay = msg.removeNextDelay();   // 用掉下一档

            // 将消息重新发回延时交换机，再延迟一档
            rabbitTemplate.convertAndSend(
                    MqConstant.DELAY_EXCHANGE,
                    MqConstant.DELAY_ORDER_ROUTING_KEY,
                    msg,
                    new MessagePostProcessor() {
                        @Override
                        public Message postProcessMessage(Message message) throws AmqpException {
                            message.getMessageProperties()
                                    .getHeaders()
                                    .put("x-delay", delay);
                            return message;
                        }
                    });
        } else {
            // 4. 所有延迟档耗尽，订单仍未支付 → 取消订单
            orderService.cancelOrder(orderNo);
        }
    }
}

消费者三分支逻辑

flowchart TD
    M[延时消息到期] --> Q{订单状态?}
    Q -->|已支付 / 订单不存在| D[return 丢弃消息]
    Q -->|仍未支付| H{还有延迟档?}
    H -->|有| R[取下一档 x-delay<br/>重新投递]
    H -->|无| C[cancelOrder 取消订单]
    R -.延迟到期.-> M

取消订单的实现

OrderServiceImpl.java

@Override
public void cancelOrder(Long orderNo) {
    UpdateWrapper<Order> updateWrapper = new UpdateWrapper<>();
    updateWrapper.eq("order_no", orderNo);
    updateWrapper.set("status", OrderStatus.CANCELLED.getCode());   // 0 = 已取消
    orderMapper.update(updateWrapper);
}

---

十一、完整执行时序

时刻	事件
T0	用户下单，order / order_item 落库（状态 = 1 未支付）；发送延时消息，x-delay = 10s，剩余梯度 [10,10,15,15,30,30]
T0 + 10s	消费：仍未支付、还有梯度 → 取下一档 10s，重新投递
T0 + 20s	仍未支付 → 再延迟 10s
T0 + 30s	仍未支付 → 再延迟 15s
…	继续按 15s、30s、30s 逐级检查
中途支付	某一档检查到状态 = 已支付 → return，消息链终止
始终未支付	最后一档到期，hasNextDelay() = false → cancelOrder() 取消订单

延迟梯度合计：10 + 10 + 10 + 15 + 15 + 30 + 30 = 120 秒（首跳 10s 由生产者设置，其余在消费时逐级触发）。

---

十二、关键设计点

1. 异步解耦：下单与超时关单互不阻塞，无需轮询订单表；
2. 精确计时、压力小：延迟计时由 MQ 插件负责，到期才触发；
3. 多级检查：用户支付后消息链自动终止，避免误关已支付订单；
4. 自动声明拓扑：@QueueBinding 启动即创建交换机、队列与绑定，无需手动在控制台配置；
5. 容错订单缺失：消费者对 order == null 直接返回，避免延时消息在数据库事务回滚后报错。

---

十三、不足与生产改进建议

问题	现状	改进建议
插件依赖	需安装延时插件，极高堆积场景性能受限	超大规模可改用 RocketMQ 延时消息或 TTL + DLX
发布确认	无 publisher confirm，broker 异常可能丢消息	开启 publisher-confirm-type: correlated 与 publisher-returns: true
消费确认	默认自动 ack、单消费者，处理异常不会重试	改为手动 ack，配置 concurrency，引入死信队列 DLX/DLQ
事务一致性	消息在 @Transactional 内发送，极端情况下事务回滚但消息已发	使用本地消息表 / Outbox，配合定时对账补偿
消费幂等	当前按状态判断，重复消息理论上需防重	使用 orderNo / 唯一 ID 做幂等去重

推荐的生产配置示例

spring:
  rabbitmq:
    publisher-confirm-type: correlated   # broker 接收确认
    publisher-returns: true              # 路由不到队列时退回
    listener:
      simple:
        acknowledge-mode: manual         # 手动 ack
        concurrency: 5                   # 消费者最小并发
        max-concurrency: 20              # 消费者最大并发
        prefetch: 10                     # 每个消费者预取消息数

---

十四、相关文件清单

文件	作用
mall-order/pom.xml	引入 spring-boot-starter-amqp
mall-order/src/main/resources/application.yml	RabbitMQ 连接配置
mall-order/.../config/RabbitMQConfig.java	JSON 消息转换器
mall-order/.../constant/MqConstant.java	交换机 / 队列 / 路由键常量
mall-order/.../util/MultiDelayMessage.java	延时消息体（订单号 + 延迟梯度）
mall-order/.../service/impl/OrderServiceImpl.java	生产者（下单发消息）、cancelOrder()
mall-order/.../listener/OrderStatusListener.java	消费者（检查 / 重投 / 取消）

---

总结

项目在 mall-order 模块内基于 Spring AMQP + RabbitMQ 延时消息插件 实现”超时未支付自动关单”：

• 下单后由 OrderServiceImpl 发送 MultiDelayMessage<Long>（订单号 + 多级延迟表），通过消息头 x-delay 交给延时交换机 trade.delay.topic；
• OrderStatusListener 通过 @RabbitListener + @QueueBinding 自动声明队列并消费：
• 已支付 / 订单不存在 → 丢弃消息；
• 未支付且有剩余延迟档 → 用下一档延迟重新投递；
• 延迟耗尽仍未支付 → cancelOrder() 取消订单。

从而在不轮询数据库的前提下，可靠、及时地完成未支付订单的超时自动取消。