RocketMQ 延时消息实现订单超时自动关闭：从方案选型到秒级精度的进阶之路

用户抢票成功 → 创建待支付订单（锁定名额）→ 用户支付 → 订单完成
                     │
                     │ 15分钟未支付
                     ▼
               自动关闭订单 → 释放名额（Redis库存回补 + MySQL库存回补）

1s  5s  10s  30s  1m  2m  3m  4m  5m  6m  7m  8m  9m  10m  20m  30m  1h  2h

// 生产者：订单创建后发送延时消息
@Service
public class OrderTimeoutProducer {
​
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
​
    public void sendTimeoutMessage(Long orderId, Long activityId, Long userId) {
        OrderTimeoutMessage msg = new OrderTimeoutMessage();
        msg.setOrderId(orderId);
        msg.setActivityId(activityId);
        msg.setUserId(userId);
        msg.setCreateTime(System.currentTimeMillis());
​
        // 延时等级15 = 20分钟
        rocketMQTemplate.syncSend("order-timeout-topic",
            MessageBuilder.withPayload(msg).build(),
            3000,    // 发送超时
            15       // 延时等级
        );
    }
}

// 消费者：延时消息到达后处理
@RocketMQMessageListener(
    topic = "order-timeout-topic",
    consumerGroup = "order-timeout-consumer"
)
@Service
public class OrderTimeoutConsumer implements RocketMQListener<OrderTimeoutMessage> {
​
    @Autowired
    private OrderService orderService;
​
    @Override
    public void onMessage(OrderTimeoutMessage msg) {
        // 查询订单当前状态
        Order order = orderService.getById(msg.getOrderId());
​
        if (order == null) {
            log.warn("订单 {} 不存在，忽略超时消息", msg.getOrderId());
            return;
        }
​
        // 如果已支付或已取消，无需处理
        if (order.getStatus() != OrderStatus.PENDING) {
            log.info("订单 {} 状态为 {}，无需关闭", msg.getOrderId(), order.getStatus());
            return;
        }
​
        // 二次确认：检查是否真的超过15分钟
        long elapsed = System.currentTimeMillis() - order.getCreateTime().getTime();
        if (elapsed < 15 * 60 * 1000) {
            // 还没到15分钟（理论上不会走到这里，但做防御性判断）
            log.info("订单 {} 未超时 ({}ms)，跳过", msg.getOrderId(), elapsed);
            return;
        }
​
        // 执行订单关闭
        orderService.closeTimeoutOrder(msg.getOrderId(), msg.getActivityId(), msg.getUserId());
    }
}

@Service
public class OrderService {
​
    @Transactional
    public void closeTimeoutOrder(Long orderId, Long activityId, Long userId) {
        // 1. 更新订单状态（乐观锁：只有PENDING状态才能关闭）
        int affected = orderMapper.closeOrder(orderId, OrderStatus.PENDING, OrderStatus.TIMEOUT_CLOSED);
        if (affected == 0) {
            // 状态已变更（可能刚刚支付成功），不处理
            log.info("订单 {} 关闭失败（状态已变更），跳过", orderId);
            return;
        }
​
        // 2. MySQL 库存回补
        activityMapper.incrementStock(activityId);
​
        // 3. Redis 库存回补
        String stockKey = "activity:{" + activityId + "}:stock";
        redisTemplate.opsForValue().increment(stockKey);
​
        // 4. Redis 已抢用户集合移除
        String usersKey = "activity:{" + activityId + "}:users";
        redisTemplate.opsForSet().remove(usersKey, userId.toString());
​
        // 5. 通知用户
        notifyService.sendOrderClosed(userId, orderId, "订单已超时关闭，名额已释放");
​
        log.info("订单 {} 超时关闭成功，活动 {} 库存已回补", orderId, activityId);
    }
}

第一条消息：延时10分钟（等级14）
    → 消费后检查：如果订单已支付则结束；如果未支付，发送第二条消息
第二条消息：延时5分钟（等级13）
    → 消费后检查并关闭订单

@RocketMQMessageListener(
    topic = "order-timeout-topic",
    consumerGroup = "order-timeout-consumer"
)
@Service
public class OrderTimeoutConsumer implements RocketMQListener<OrderTimeoutMessage> {
​
    @Override
    public void onMessage(OrderTimeoutMessage msg) {
        Order order = orderService.getById(msg.getOrderId());
​
        if (order == null || order.getStatus() != OrderStatus.PENDING) {
            return; // 已支付或已取消，无需处理
        }
​
        if (msg.getPhase() == 1) {
            // 第一阶段消息到达（10分钟后），订单仍未支付
            // 发送第二阶段延时消息（再等5分钟）
            msg.setPhase(2);
            rocketMQTemplate.syncSend("order-timeout-topic",
                MessageBuilder.withPayload(msg).build(),
                3000, 13); // 等级13 = 5分钟
​
            // 可选：给用户发送"即将关闭"的提醒
            notifyService.sendPaymentReminder(order.getUserId(), order.getOrderId(),
                "您的订单将在5分钟后关闭，请尽快完成支付");
​
        } else if (msg.getPhase() == 2) {
            // 第二阶段消息到达（总计15分钟后），执行关闭
            orderService.closeTimeoutOrder(msg.getOrderId(), msg.getActivityId(), msg.getUserId());
        }
    }
}

@Scheduled(fixedRate = 300000) // 每5分钟
public void scanTimeoutOrders() {
    // 查找超过20分钟仍未支付的订单（比正常超时多5分钟的安全余量）
    LocalDateTime deadline = LocalDateTime.now().minusMinutes(20);
    List<Order> timeoutOrders = orderMapper.selectTimeoutOrders(
        OrderStatus.PENDING, deadline, 100); // 每次最多处理100条
​
    if (!timeoutOrders.isEmpty()) {
        log.warn("兜底扫描发现 {} 条超时未关闭订单", timeoutOrders.size());
        for (Order order : timeoutOrders) {
            try {
                orderService.closeTimeoutOrder(
                    order.getId(), order.getActivityId(), order.getUserId());
            } catch (Exception e) {
                log.error("兜底关闭订单 {} 失败", order.getId(), e);
            }
        }
    }
}