RabbitMQ消息队列：延迟消息-北京尧图网络

一、方案一死信交换机 TTLTime-To-LiveRabbitMQ本身并没有直接提供延迟消息的功能但我们可以巧妙地利用死信交换机Dead Letter Exchange和消息TTL过期时间来模拟实现。1.1 什么是死信交换机当一个消息在一个队列中变为“死信”时它会被重新投递到指定的死信交换机再由它路由到最终的队列。消息成为死信的三种情况消费者拒绝消费使用basic.reject或basic.nack声明消费失败且requeue参数设为false。消息过期消息在队列中存活时间超过了设置的 TTL。队列达到最大长度队列满了无法再接纳新消息。当一个队列配置了dead-letter-exchange属性那么发生上述情况的消息就会被转发到该交换机。1.2 利用死信交换机实现延迟消息的核心思想消息投递生产者将消息发送到一个没有消费者的普通队列并设置消息的 TTL例如5秒。消息过期消息在队列中存活到 TTL 结束后变为死信。死信转发该队列配置了死信交换机因此死信被转发到死信交换机。最终消费死信交换机根据路由规则将消息投递到最终的业务队列由消费者处理。此时从消息发送到消费者收到刚好经历了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类其中包含业务数据和一个ListLong类型的延迟时间集合单位毫秒。Data public class MultiDelayMessageT { private T data; private ListLong delayMillis; // 获取并移除第一个延迟时间实现“消费一个取一个”的效果 public Long removeNextDelay(){ return delayMillis.remove(0); } public boolean hasNextDelay(){ return !delayMillis.isEmpty(); } }3.2.2 服务改造与配置定义常量明确交换机、队列、路由Key。public interface MqConstants { String DELAY_EXCHANGE trade.delay.topic; String DELAY_ORDER_QUEUE trade.order.delay.queue; String DELAY_ORDER_ROUTING_KEY order.query; }引入依赖在交易服务中引入 Spring AMQP 依赖。共享MQ配置将 RabbitMQ 的连接信息抽取到 Nacos 配置中心方便统一管理。3.2.3 改造下单业务在用户下单成功后立即发送第一条延迟消息例如10秒后。// 创建订单后... // 发送延迟消息检查支付状态 // 延迟时间数组10秒、30秒、1分钟... MultiDelayMessageLong 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 核心监听器处理逻辑消息监听器是整个流程的大脑其处理逻辑如下消费消息从delay.queue获取包含订单ID的延迟消息。检查本地订单状态若订单已支付或已关闭直接结束。查询支付服务若本地订单仍为“未支付”则远程调用支付服务查询最新状态。状态判断已支付更新本地订单状态为“已支付”流程结束。未支付判断MultiDelayMessage中是否还有剩余延迟时间。有取出下一个延迟时间重新发送延迟消息。无说明已超过最大等待时间如30分钟执行业务取消订单、恢复库存。javaRabbitListener(bindings QueueBinding(...)) public void listenOrderCheckDelayMessage(MultiDelayMessageLong msg) { // 1. 获取订单ID // 2. 本地订单状态检查 // 3. 远程查询支付状态 // 4. 支付成功更新订单 // 5. 未支付判断是否继续延迟检测 if (msg.hasNextDelay()) { int delayVal msg.removeNextDelay().intValue(); // 重新发送延迟消息x-delay delayVal } else { // 6. 超时未支付取消订单 orderService.cancelOrder(orderId); } }四、总结本文详细介绍了RabbitMQ实现延迟消息的两种主流方案并深入讲解了其在电商订单超时处理场景下的实战应用。方案实现方式优点缺点适用场景死信交换机 TTL利用消息过期和死信转发机制无需额外插件基于核心功能配置复杂延迟时间可能不精确对时间精度要求不高且不想引入插件的场景DelayExchange 插件使用官方插件设置x-delay属性使用简单延迟精度高需要安装插件大量长延迟消息有性能开销对时间精度有要求且延迟时间不宜过长的场景。生产环境更推荐关键点回顾延迟消息是解决分布式系统中定时任务的一种优雅方案。“梯度延迟检测”策略能有效降低MQ资源消耗是优化延迟任务的重要手段。结合Feign 远程调用与RabbitMQ可以实现服务间的松耦合和高效协作。

RabbitMQ消息队列：延迟消息

相关推荐

Unity多人游戏开发利器：ParrelSync本地多开调试实战指南

League-Toolkit：基于LCU API的英雄联盟自动化工具技术解析

用友ERP系统技术架构与实施实战指南

更多资讯

从汽车到具身智能：全域安全迁移的底层逻辑与实践路径

SpringBoot+Vue房屋租赁系统开发实战与优化

dnf85元素刷图加点实战：搞定高频面试题与环境配置痛点

DOM Element核心概念与前端开发实践指南

5个工具搞定生日歌曲下载，保姆级教程避坑指南

3步搞定小娜怎么关闭，程序员从入门到精通避坑指南

今日更新

OneUptime 自定义探针（Custom Probe）部署实战：私网监控、代理配置与断连排障全指南

大众TL52625前端框架材料要求详解：从性能测试到落地执行

TiXL 浮点运算算子库 Lib.numbers.float 完全指南：44 个算子的参数详解、源码原理与实战串联

本周精选

深入解析Transformer多头注意力机制与工程优化

OpenClaw 的 Skills 跑学习任务，模型通道改到 TaoToken 通道行不行？

ChatGPT报错Oops, an error occurred! 全链路排查指南

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？