使用RabbitMQ实现延迟消息的完整指南
在分布式系统中,消息队列通常用于解耦服务,RabbitMQ是一个广泛使用的消息队列服务。延迟消息(也称为延时队列或TTL消息)是一种常见的场景应用,特别适合处理某些任务在一段时间后执行的需求,如订单超时处理、延时通知等。
本文将以具体代码为例,展示如何使用RabbitMQ来实现延迟消息处理,涵盖队列和交换机的配置、消息的发送与接收以及死信队列的处理。
什么是延迟消息?
延迟消息是指消息在发送到队列后,经过设定的时间延迟再被消费。RabbitMQ 本身没有直接支持延迟队列的功能,但可以通过 TTL(Time To Live)+ 死信队列(Dead Letter Queue, DLQ) 的组合来实现。当消息超过TTL(消息存活时间)后,不会被立即消费,而是会被转发到绑定的死信队列,从而实现延迟处理。
RabbitMQ中的延迟消息原理
在RabbitMQ中,我们可以通过以下几个概念来实现延迟消息:
- TTL(Time To Live):可以为队列设置TTL,消息超过该时间后会被标记为“死信”。
- 死信队列(Dead Letter Queue):当消息在正常队列中过期或处理失败时,RabbitMQ可以将它们路由到一个死信队列,死信队列可以用来处理这些过期或未处理的消息。
- x-dead-letter-exchange 和 x-dead-letter-routing-key:可以通过配置队列的参数,将过期消息发送到一个专门的死信交换器,并根据指定的路由键转发到死信队列。

消息来到ttl.queue消息队列,过期时间内无人消费,消息来到死信交换机hmall.direct,在direct.queue消息队列无需等待。
1. RabbitMQ的配置
首先,我们需要配置两个队列和两个交换机:一个用于存放延时消息,另一个用于处理超时的死信消息。
package com.heima.stroke.configuration;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;@Configuration
public class RabbitConfig {// 延迟时间 单位:毫秒 (这里设为30秒)private static final long DELAY_TIME = 1000 * 30;// 行程超时队列public static final String STROKE_OVER_QUEUE = "STROKE_OVER_QUEUE";// 行程死信队列public static final String STROKE_DEAD_QUEUE = "STROKE_DEAD_QUEUE";// 行程超时队列交换机public static final String STROKE_OVER_QUEUE_EXCHANGE = "STROKE_OVER_QUEUE_EXCHANGE";// 行程死信队列交换机public static final String STROKE_DEAD_QUEUE_EXCHANGE = "STROKE_DEAD_QUEUE_EXCHANGE";// 行程超时交换机 Routing Keypublic static final String STROKE_OVER_KEY = "STROKE_OVER_KEY";// 行程死信交换机 Routing Keypublic static final String STROKE_DEAD_KEY = "STROKE_DEAD_KEY";/*** 声明行程超时队列,并设置其参数* x-dead-letter-exchange:绑定的死信交换机* x-dead-letter-routing-key:死信路由Key* x-message-ttl:消息的过期时间*/@Beanpublic Queue strokeOverQueue() {Map<String, Object> args = new HashMap<>(3);args.put("x-dead-letter-exchange", STROKE_DEAD_QUEUE_EXCHANGE);args.put("x-dead-letter-routing-key", STROKE_DEAD_KEY);args.put("x-message-ttl", DELAY_TIME); // 设置TTL为30秒return QueueBuilder.durable(STROKE_OVER_QUEUE).withArguments(args).build();}@Beanpublic DirectExchange strokeOverQueueExchange() {return new DirectExchange(STROKE_OVER_QUEUE_EXCHANGE);}@Beanpublic Binding bindingStrokeOverDirect() {return BindingBuilder.bind(strokeOverQueue()).to(strokeOverQueueExchange()).with(STROKE_OVER_KEY);}
}
解释:
TTL设置:我们通过x-message-ttl设置消息的过期时间为30秒。
死信队列绑定:通过x-dead-letter-exchange和x-dead-letter-routing-key设置,当消息过期时,它会被转发到死信交换机,再路由到死信队列。
2. 生产者发送延迟消息
接下来,我们通过生产者向超时队列发送消息,这些消息将在TTL过期后转发到死信队列。
package com.heima.stroke.rabbitmq;import com.alibaba.fastjson.JSON;
import com.heima.modules.vo.StrokeVO;
import com.heima.stroke.configuration.RabbitConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Component
public class MQProducer {private final static Logger logger = LoggerFactory.getLogger(MQProducer.class);@AutowiredRabbitTemplate rabbitTemplate;/*** 发送延时消息到行程超时队列** @param strokeVO 消息体*/public void sendOver(StrokeVO strokeVO) {String mqMessage = JSON.toJSONString(strokeVO);logger.info("send timeout msg:{}", mqMessage);rabbitTemplate.convertAndSend(RabbitConfig.STROKE_OVER_QUEUE_EXCHANGE, RabbitConfig.STROKE_OVER_KEY, mqMessage);}
}
解释:
sendOver 方法将消息发送到超时队列,消息将在超时后进入死信队列。生产者不需要额外处理TTL或死信的配置,只需发送消息即可。
3. 消费者监听死信队列
当消息超过TTL后,将会被转发到死信队列。消费者需要监听死信队列并处理这些消息。
j
package com.heima.stroke.rabbitmq;import com.alibaba.fastjson.JSON;
import com.heima.modules.vo.StrokeVO;
import com.heima.stroke.configuration.RabbitConfig;
import com.heima.stroke.handler.StrokeHandler;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;@Component
public class MQConsumer {private final static Logger logger = LoggerFactory.getLogger(MQConsumer.class);@Autowiredprivate StrokeHandler strokeHandler;/*** 监听死信队列** @param message 消息体* @param channel RabbitMQ的Channel* @param tag 消息的Delivery Tag*/@RabbitListener(bindings = {@QueueBinding(value = @Queue(value = RabbitConfig.STROKE_DEAD_QUEUE, durable = "true"),exchange = @Exchange(value = RabbitConfig.STROKE_DEAD_QUEUE_EXCHANGE),key = RabbitConfig.STROKE_DEAD_KEY)})@RabbitHandlerpublic void processStroke(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {StrokeVO strokeVO = JSON.parseObject(message.getBody(), StrokeVO.class);logger.info("get dead msg:{}", message.getBody());if (strokeVO == null) {return;}try {// 处理超时的行程消息strokeHandler.timeoutHandel(strokeVO);// 手动确认消息channel.basicAck(tag, false);} catch (Exception e) {e.printStackTrace();}}
}
解释:
@RabbitListener 注解绑定了死信队列的监听器。当消息被转发到死信队列时,该消费者会接收到消息。
使用 channel.basicAck(tag, false) 手动确认消息处理成功,确保消息不会重复消费。
4. 处理超时业务逻辑
在我们的业务中,当消息超时未处理时,将其状态设置为超时。
public void timeoutHandel(StrokeVO strokeVO) {// 获取司机行程ID和乘客行程IDString inviterTripId = strokeVO.getInviterTripId();String inviteeTripId = strokeVO.getInviteeTripId();// 检查邀请状态是否为未确认String inviteeStatus = redisHelper.getHash(HtichConstants.STROKE_INVITE_PREFIX, inviteeTripId, inviterTripId);String inviterStatus = redisHelper.getHash(HtichConstants.STROKE_INVITE_PREFIX, inviterTripId, inviteeTripId);if (String.valueOf(InviteState.UNCONFIRMED.getCode()).equals(inviteeStatus) &&String.valueOf(InviteState.UNCONFIRMED.getCode()).equals(inviterStatus)) {// 更新为超时状态redisHelper.addHash(HtichConstants.STROKE_INVITE_PREFIX, inviteeTripId, inviterTripId, String.valueOf(InviteState.TIMEOUT.getCode()));redisHelper.addHash(HtichConstants.STROKE_INVITE_PREFIX, inviterTripId, inviteeTripId, String.valueOf(InviteState.TIMEOUT.getCode()));}
}
相关文章:
使用RabbitMQ实现延迟消息的完整指南
在分布式系统中,消息队列通常用于解耦服务,RabbitMQ是一个广泛使用的消息队列服务。延迟消息(也称为延时队列或TTL消息)是一种常见的场景应用,特别适合处理某些任务在一段时间后执行的需求,如订单超时处理、…...
阿里员工:阿里工作7年至少得P7吧,快的都P8了,年薪100W是正常的,80才算及格...
上一篇:一线体面男的收入 年薪64W的阿里蚂蚁员工爆料:在阿里,工作7年至少得P7,快的都P8了,年薪100W才正常,80分才算及格。 其实,在大厂工作,听起来风光无限,但个中滋味&a…...
Django进一步掌握(10月22日)
一、请求响应对象 请求对象request 响应对象HttpResponse 二、HttpResponse常用属性 status设置HTTP响应状态码 status_code查询HTTP响应状态码 content_type设置响应的类型 write()写入响应内容 三、重定向 1、实现URl访问的重定向 (1)使用Ht…...
C++从入门到起飞之——红黑树封装map和set 全方位剖析!
目录 1、map和set的整体框架 2、map和set迭代器的实现 3、map支持[] 4、完整源码 set.h map.h RBTree.h 1、map和set的整体框架 因为map和set的底层都是红黑树,所以我们考虑用一个红黑树的类模版去实例化map和set对象!不过,map节点中存…...
【javax maven项目缺少_Maven的依赖管理 引入依赖】
javax maven项目缺少_Maven的依赖管理 引入依赖 Maven的依赖管理 - 引入依赖依赖管理(引入依赖)导入依赖 https://blog.csdn.net/weixin_28932089/article/details/112381468 Maven的依赖管理 - 引入依赖 依赖管理(引入依赖) 能够掌握依赖引入的配置方式 导入依赖 导入依赖练…...
手搓一个定时器
目录 1.什么是定时器 2.计时器的使用 3.手搓定时器 3.1定义一个TimerTask类 3.2定义一个Timer类 3.3实现schedule方法 3.4实现Timer的构造方法 3.4.1随时随地查看优先级队列中是否有任务要执行 3.4.2获取队首任务,并判断是否到执行时间 3.4.3到达执行时间…...
AI提示词工程优化Prompt-GPT使用手册(科普一键收藏史上最强攻略)
Prompt(提示),最初是 NLP 研究者为下游任务设计出来的一种任务专属的输入形式或模板。在 ChatGPT 引发大语言模型新时代之后,Prompt 指与大模型交互输入的代称。 随着大模型的进展,Prompt Engineering是一个持久的探索过程。 目录 什么是提示…...
【数据结构】快速排序(三种实现方式)
目录 一、基本思想 二、动图演示(hoare版) 三、思路分析(图文) 四、代码实现(hoare版) 五、易错提醒 六、相遇场景分析 6.1 ❥ 相遇位置一定比key要小的原因 6.2 ❥ 右边为key,左边先走 …...
利用前向勾子获取神经网络中间层的输出并将其进行保存(示例详解)
代码示例: # 激活字典,用于保存每次的中间特征 activation {}# 将 forward_hook 函数定义在 upsample_v2 外部 def forward_hook(name):def hook(module, input, output):activation[name] output.detach()return hookdef upsample_v2(in_channels, o…...
CTF-RE 从0到N: S盒
S盒(Substitution Box) 是密码学中的一种替换表,用于对输入数据进行非线性变换,以增加加密过程的复杂性。它主要用于对称加密算法中(例如AES、DES),作为加密轮次的一部分,对输入字节…...
MT-Pref数据集:包含18种语言的18k实例,涵盖多个领域。实验表明它能有效提升Tower模型在WMT23和FLORES基准测试中的翻译质量。
2024-10-10,由电信研究所、里斯本大学等联合创建MT-Pref数据集,它包含18种语言方向的18k实例,覆盖了2022年后的多个领域文本。通过在WMT23和FLORES基准测试上的实验,我们展示了使用MT-Pref数据集对Tower模型进行对齐可以显著提高翻…...
【C++ 真题】B2099 矩阵交换行
矩阵交换行 题目描述 给定一个 5 5 5 \times 5 55 的矩阵(数学上,一个 r c r \times c rc 的矩阵是一个由 r r r 行 c c c 列元素排列成的矩形阵列),将第 n n n 行和第 m m m 行交换,输出交换后的结果。 输入格式 输入共 6 6 6 …...
AAPL: Adding Attributes to Prompt Learning for Vision-Language Models
文章汇总 当前的问题 1.元标记未能捕获分类的关键语义特征 如下图(a)所示, π \pi π在类聚类方面没有显示出很大的差异,这表明元标记 π \pi π未能捕获分类的关键语义特征。我们进行简单的数据增强后,如图(b)所示,效果也是如…...
MySQLDBA修炼之道-开发篇(一)
三、开发基础 1. 数据模型 1.1 关系数据模型介绍 关于NULL 如果某个字段的值是未知的或未定义的,数据库会提供一个特殊的值NULL来表示。NULL值很特殊,在关系数据库中应该小心处理。例如查询语句“select*from employee where 绩效得分<85 or>绩…...
Spring MVC 知识点全解析
Spring MVC 知识点全解析 Spring MVC 是一个基于 Java 的请求驱动的 Web 框架,属于 Spring 框架的一部分,广泛用于构建企业级 Web 应用程序。本文将详细阐述 Spring MVC 的核心知识点,包括其工作原理、关键组件、配置、请求处理、数据绑定、…...
python 基于FastAPI实现一个简易的在线用户统计 服务
简易在线用户统计服务 概述 这是一个基于Python的FastAPI框架实现的服务,用于统计客户端的心跳信息,并据此维护在线用户列表以及记录活跃用户数。 功能特性 心跳接收:接受来自客户端的心跳包,以更新客户端的状态。在线用户统计…...
glibc中xdr的一个bug
本人在64位linux服务器上(centos7),发现xdr_u_long这个函数有个bug,就是数字的范围如果超过unsigned int的最大值(4294967295)时,xdr_u_long失败。 这个场景主要用在unix时间戳上面,比如一款软件,设置有效期为100年。…...
Android Framework定制sim卡插入解锁pin码的界面
文章目录 手机设置SIM卡pin码一、安卓手机二、苹果手机 Android Framework中SIM卡pin码代码定位pin码提示文本位置定位pin码java代码位置 定制pin码framework窗口数字按钮 手机设置SIM卡pin码 设置 SIM 卡 PIN 码可以提高手机的安全性,防止他人在未经授权的情况下使…...
cc2530 Basic RF 讲解 和点灯讲解(1_1)
1. Basic RF 概述 Basic RF 是 TI 提供的一套简化版的无线通信协议栈,旨在帮助开发者快速搭建无线通信系统。它基于 IEEE 802.15.4 标准的数据包收发,但只用于演示无线设备数据传输的基本方法,不包含完整功能的协议。Basic RF 的功能限制包括…...
Android H5页面性能分析策略
文章目录 引言一、拦截资源加载请求以优化性能二、通过JavaScript代码监控资源下载速度三、使用vConsole进行前端性能调试四、使用Chrome DevTools调试Android端五、通过抓包分析优化网络性能六、总结 引言 在移动应用开发中,H5页面的性能直接影响到用户体验。本文…...
树莓派超全系列教程文档--(62)使用rpicam-app通过网络流式传输视频
使用rpicam-app通过网络流式传输视频 使用 rpicam-app 通过网络流式传输视频UDPTCPRTSPlibavGStreamerRTPlibcamerasrc GStreamer 元素 文章来源: http://raspberry.dns8844.cn/documentation 原文网址 使用 rpicam-app 通过网络流式传输视频 本节介绍来自 rpica…...
Spring Boot 实现流式响应(兼容 2.7.x)
在实际开发中,我们可能会遇到一些流式数据处理的场景,比如接收来自上游接口的 Server-Sent Events(SSE) 或 流式 JSON 内容,并将其原样中转给前端页面或客户端。这种情况下,传统的 RestTemplate 缓存机制会…...
iPhone密码忘记了办?iPhoneUnlocker,iPhone解锁工具Aiseesoft iPhone Unlocker 高级注册版分享
平时用 iPhone 的时候,难免会碰到解锁的麻烦事。比如密码忘了、人脸识别 / 指纹识别突然不灵,或者买了二手 iPhone 却被原来的 iCloud 账号锁住,这时候就需要靠谱的解锁工具来帮忙了。Aiseesoft iPhone Unlocker 就是专门解决这些问题的软件&…...
vue3 字体颜色设置的多种方式
在Vue 3中设置字体颜色可以通过多种方式实现,这取决于你是想在组件内部直接设置,还是在CSS/SCSS/LESS等样式文件中定义。以下是几种常见的方法: 1. 内联样式 你可以直接在模板中使用style绑定来设置字体颜色。 <template><div :s…...
【ROS】Nav2源码之nav2_behavior_tree-行为树节点列表
1、行为树节点分类 在 Nav2(Navigation2)的行为树框架中,行为树节点插件按照功能分为 Action(动作节点)、Condition(条件节点)、Control(控制节点) 和 Decorator(装饰节点) 四类。 1.1 动作节点 Action 执行具体的机器人操作或任务,直接与硬件、传感器或外部系统…...
Spring Boot+Neo4j知识图谱实战:3步搭建智能关系网络!
一、引言 在数据驱动的背景下,知识图谱凭借其高效的信息组织能力,正逐步成为各行业应用的关键技术。本文聚焦 Spring Boot与Neo4j图数据库的技术结合,探讨知识图谱开发的实现细节,帮助读者掌握该技术栈在实际项目中的落地方法。 …...
华为云Flexus+DeepSeek征文|DeepSeek-V3/R1 商用服务开通全流程与本地部署搭建
华为云FlexusDeepSeek征文|DeepSeek-V3/R1 商用服务开通全流程与本地部署搭建 前言 如今大模型其性能出色,华为云 ModelArts Studio_MaaS大模型即服务平台华为云内置了大模型,能助力我们轻松驾驭 DeepSeek-V3/R1,本文中将分享如何…...
2023赣州旅游投资集团
单选题 1.“不登高山,不知天之高也;不临深溪,不知地之厚也。”这句话说明_____。 A、人的意识具有创造性 B、人的认识是独立于实践之外的 C、实践在认识过程中具有决定作用 D、人的一切知识都是从直接经验中获得的 参考答案: C 本题解…...
初学 pytest 记录
安装 pip install pytest用例可以是函数也可以是类中的方法 def test_func():print()class TestAdd: # def __init__(self): 在 pytest 中不可以使用__init__方法 # self.cc 12345 pytest.mark.api def test_str(self):res add(1, 2)assert res 12def test_int(self):r…...
高效线程安全的单例模式:Python 中的懒加载与自定义初始化参数
高效线程安全的单例模式:Python 中的懒加载与自定义初始化参数 在软件开发中,单例模式(Singleton Pattern)是一种常见的设计模式,确保一个类仅有一个实例,并提供一个全局访问点。在多线程环境下,实现单例模式时需要注意线程安全问题,以防止多个线程同时创建实例,导致…...
