当前位置：首页 > news >正文

RabbitMQ——延迟消息的实现

news 来源：原创 2025/6/18 8:03:34

延迟消息功能在消息队列（MQ）中占据重要地位，它允许消息在指定时间后才被消费者处理，适用于需要定时任务或等待特定条件满足的场景。企业利用这一功能可以优化业务流程，如订单超时取消、定时促销等，确保系统响应更加灵活高效。通过延迟消息，企业不仅能提升用户体验，还能有效管理资源，避免因即时处理所有消息而导致的系统过载，增强了系统的可靠性和可扩展性。

问题：虽然我们利用各种机制尽可能增加了消息的可靠性，但是MQ通知依旧可能失败。

解决方案：既然MQ通知不一定发送到交易服务，那么交易服务就必须自己主动去查询支付状态。这样即便支付服务的MQ通知失败，我们依然能通过主动查询来保证订单状态的一致。

场景描绘：

用户小李正在一家在线票务平台上预订一张热门演唱会的门票。 在选择好心仪的座位并确认订单后，系统立即锁定了该座位，确保没有其他用户可以同时购买同一张票。此时，页面提示小李需要在接下来的30分钟内完成支付，否则订单将自动取消，座位也将重新释放给其他用户。

后台系统启动了一个定时任务，专门用于监控所有未支付订单的状态。 这个定时任务每分钟都会检查是否有订单接近其支付超时时间点。当检测到小李的订单距离超时只剩下最后几秒钟而仍未完成支付时，系统开始执行一系列预定的操作。

首先，系统会尝试最后一次查询小李的支付状态，以确保没有任何延迟的通知遗漏。如果确认支付确实未完成，那么系统将自动取消该订单，并更新数据库中的订单状态为“已取消”。与此同时，之前被锁定的座位资源也会被立即释放，使其他感兴趣的顾客有机会购买这张票。

问题：如何才能准确的实现在下单后第30分钟去检查支付状态呢？

解决方案：RabbitMQ中实现延迟消息

在RabbitMQ中实现延迟消息也有两种方案：

1.死信交换机+TTL

2.延迟消息插件

什么是死信

当一个队列中的消息满足下列情况之一时，可以成为死信（dead letter）：

1.消费者使用basic.reject或 basic.nack声明消费失败，并且消息的requeue参数设置为false

2.消息是一个过期消息，超时无人消费

3.要投递的队列消息满了，无法投递

什么是死信交换机

在消息队列系统中，当一条消息由于未被成功消费而成为死信，并且原队列配置了dead-letter-exchange属性指向特定的交换机时，这个交换机就被称为死信交换机（Dead Letter Exchange, DLX）。随后，这些死信消息会被投递到DLX，并根据其绑定规则路由到与之关联的队列中。如果有队列通过绑定键与该DLX相连，则死信消息最终会被投递到这个队列，供专门的消费者进行进一步处理。

工作流程：

基于死信队列虽然可以实现延迟消息，但是太繁琐了。

所以我推荐使用方案二：延迟消息插件

插件下载地址：

https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq

基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。

docker volume inspect mq-plugins

接下来执行命令，安装插件：

cd /var/lib/docker/volumes/mq-plugins/_data // 进入目录
docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

声明延迟交换机

基于注解方式：

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "delay.queue"),
            exchange = @Exchange(name = "delay.exchange", delayed = "true"),
            key = "hhh"
    ))
    public void listenDelayQueue(String msg){
        System.out.println("消费者1接收到dix.queue的消息：【" + msg + "】"+
                LocalTime.now());
    }

发送消息时，必须通过x-delay属性设定延迟时间：

@Test
void testSendDelay(){
    // 发送延迟消息到RabbitMQ
    // 使用RabbitTemplate的convertAndSend方法，指定交换机名、路由键和消息内容
    // 通过Lambda表达式设置消息属性，使其延迟5000毫秒
    rabbitTemplate.convertAndSend("delay.exchange", "hhh", "hello delay",
           message -> {
              // 设置消息的延迟时间
              message.getMessageProperties().setDelay(5000);
              // 返回设置完延迟属性的消息
              return message;
           });
}

注意：

延迟消息插件内部会维护一个本地数据库表，同时使用Elang Timers功能实现计时。如果消息的延迟时间设置较长，可能会导致堆积的延迟消息非常多，会带来较大的CPU开销，同时延迟消息的时间会存在误差。