RabbitMQ的死信队列和延迟队列

文章目录

    • 死信队列
    • 如何配置死信队列
      • 死信队列的应用场景
      • Spring Boot实现RabbitMQ的死信队列
    • 延迟队列
      • 方案优劣:
      • 延迟队列的实现有两种方式:

死信队列

1)“死信”是RabbitMQ中的一种消息机制。
2)消息变成死信,可能是由于以下的原因
● 消息被拒绝
● 消息过期
● 队列达到最大长度
3)死信队列
当消息在一个队列中变成死信(dead message)之后,它能被重新发送到另一个交换机中,这个交换机就是 DLX(Dead-Letter-Exchange ) ,绑定 DLX 的队列就称之为死信队列。
“死信”消息会被RabbitMQ进行特殊处理,如果配置了死信队列信息,那么该消息将会被丢进死信队列中,如果没有配置,则该消息将会被丢弃。

如何配置死信队列

  1. 配置业务队列,绑定到业务交换机上
  2. 为业务队列配置死信交换机和路由key
  3. 为死信交换机配置死信队列
    注意:并不是直接声明一个公共的死信队列,然后所以死信消息就自己跑到死信队列里去了。而是为每个需要使用死信的业务队列配置一个死信交换机,这里同一个项目的死信交换机可以共用一个,然后为每个业务队列分配一个单独的路由key。
    有了死信交换机和路由key后,接下来,就像配置业务队列一样,配置死信队列,然后绑定在死信交换机上。也就是说,死信队列并不是什么特殊的队列,只不过是绑定在死信交换机上的队列。死信交换机也不是什么特殊的交换机,只不过是用来接受死信的交换机,所以可以为任何类型【Direct、Fanout、Topic】。一般来说,会为每个业务队列分配一个独有的路由key,并对应的配置一个死信队列进行监听,也就是说,一般会为每个重要的业务队列配置一个死信队列。
    具体因为队列消息过期而被投递到死信队列的流程:
    在这里插入图片描述

死信队列其实并没有什么神秘的地方,不过是绑定在死信交换机上的普通队列,而死信交换机也只是一个普通的交换机,不过是用来专门处理死信的交换机。
死信消息的生命周期:
1)业务消息被投入业务队列
2)消费者消费业务队列的消息,由于处理过程中发生异常,于是进行了nck或者reject操作
3)被nck或reject的消息由RabbitMQ投递到死信交换机中
4)死信交换机将消息投入相应的死信队列
5)死信队列的消费者消费死信消息
死信消息是RabbitMQ为我们做的一层保证,其实我们也可以不使用死信队列,而是在消息消费异常时,将消息主动投递到另一个交换机中,当你明白了这些之后,这些Exchange和Queue想怎样配合就能怎么配合。比如从死信队列拉取消息,然后发送邮件、短信、钉钉通知来通知开发人员关注。或者将消息重新投递到一个队列然后设置过期时间,来进行延时消费

死信队列的应用场景

一般用在较为重要的业务队列中,确保未被正确消费的消息不被丢弃,一般发生消费异常可能原因主要有由于消息信息本身存在错误导致处理异常,处理过程中参数校验异常,或者因网络波动导致的查询异常等等,当发生异常时,当然不能每次通过日志来获取原消息,然后让运维帮忙重新投递消息。
通过配置死信队列,可以让未正确处理的消息暂存到另一个队列中,待后续排查清楚问题后,编写相应的处理代码来处理死信消息,这样比手工恢复数据要好太多了。

Spring Boot实现RabbitMQ的死信队列

当您在使用Spring Boot实现RabbitMQ的死信队列时,您需要完成以下步骤:

  1. 添加Maven依赖
    确保您的pom.xml文件中包含以下Maven依赖:
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

这将为您的应用程序提供RabbitMQ的基本支持。
2. 配置RabbitMQ连接信息
在application.properties或application.yml文件中添加RabbitMQ的连接信息,包括主机地址、用户名、密码等。
3. 创建RabbitMQConfig类
创建一个配置类,用于定义交换机、队列以及它们之间的绑定关系。在这个类中,您需要定义普通队列、死信队列、死信交换机,并将它们进行绑定。
4. 创建消费者类
创建一个消费者类,使用@RabbitListener注解标记需要监听的死信队列,并在方法上使用@RabbitHandler注解来处理接收到的死信消息。
5. 发送消息到普通队列
在需要发送消息的地方,使用RabbitTemplate发送消息到普通队列中。当消息因为过期或被拒绝接收时,会被标记为死信消息,并根据参数设置转发到死信交换机中,然后路由到死信队列中。

下面是一个简单的示例代码,演示了如何在Spring Boot中实现RabbitMQ的死信队列,并消费死信队列的消息:

// RabbitMQConfig.java
@Configuration
public class RabbitMQConfig {@Beanpublic Queue myQueue() {return QueueBuilder.durable("my_queue").withArgument("x-dead-letter-exchange", "dlx_exchange").withArgument("x-dead-letter-routing-key", "dlq_queue").build();}@Beanpublic Queue dlqQueue() {return QueueBuilder.durable("dlq_queue").build();}@Beanpublic Exchange dlxExchange() {return ExchangeBuilder.directExchange("dlx_exchange").durable(true).build();}@Beanpublic Binding binding(Queue myQueue, Exchange dlxExchange) {return BindingBuilder.bind(myQueue).to(dlxExchange).with("my_queue").noargs();}
}// DeadLetterQueueConsumer.java
@Component
@RabbitListener(queues = "dlq_queue")
public class DeadLetterQueueConsumer {@RabbitHandlerpublic void processDeadLetterMessage(String message) {System.out.println("Received message from dead letter queue: " + message);// 处理接收到的死信消息}
}// RabbitMQService.java
@Service
public class RabbitMQService {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendMessage() {rabbitTemplate.convertAndSend("my_queue", "Hello, RabbitMQ!");}
}

通过以上步骤,您就可以在Spring Boot项目中实现RabbitMQ的死信队列,并消费死信队列的消息。

延迟队列

延迟队列存储的对象是对应的延迟消息;所谓“延迟消息” 是指当消息被发送以后,并不想让消费者立刻拿到消息,而是等待特定时间后,消费者才能拿到这个消息进行消费。
在RabbitMQ中延迟队列可以通过 过期时间 + 死信队列 来实现;具体如下流程图所示:

在这里插入图片描述

RabbitMQ 的基因中没有延时队列这回事,它不能直接指定一个队列类型为延时队列,然后去延时处理,但是经过上面两节的铺垫,我们可以将 TTL+DLX 相结合,这就能组成一个延时队列。
设想一个场景,下完订单之后 15 分钟未付款我们就要将订单关闭,这就是一个很经典的演示消费的场景,如果拿 RabbitMQ 来做,我们就需要结合 TTL+DLX 了。
先把订单消息设置好 15 分钟过期时间,然后过期后队列将消息转发给我们设置好的 DLX-Exchange,DLX-Exchange 再将分发给它绑定的队列,我们的消费者再消费这个队列中的消息,就做到了延时十五分钟消费。

RabbitMQ 有两个特性,一个是 Time-To-Live Extensions,另一个是 Dead Letter Exchanges。
Time-To-Live Extensions
RabbitMQ允许我们为消息或者队列设置TTL(time to live),也就是过期时间。TTL表明了一条消息可在队列中存活的最大时间,单位为毫秒。也就是说,当某条消息被设置了TTL或者当某条消息进入了设置了TTL的队列时,这条消息会在经过TTL秒后 “死亡”,成为Dead Letter。如果既配置了消息的TTL,又配置了队列的TTL,那么较小的那个值会被取用。
Dead Letter Exchanges
在 RabbitMQ 中,一共有三种消息的 “死亡” 形式:
● 消息被拒绝。通过调用 basic.reject 或者 basic.nack 并且设置的 requeue 参数为 false;
● 消息因为设置了TTL而过期;
● 队列达到最大长度。

DLX同一般的 Exchange 没有区别,它能在任何的队列上被指定,实际上就是设置某个队列的属性。当队列中有 DLX 消息时,RabbitMQ就会自动的将 DLX 消息重新发布到设置的 Exchange 中去,进而被路由到另一个队列,publish 可以监听这个队列中消息做相应的处理。
由上简介大家可以看出,RabbitMQ本身是不支持延迟队列的,只是他的特性让勤劳的 中国脱发群体 急中生智(为了完成任务)弄出了这么一套可用的方案。
可用的方案就是:

  1. 如果有事件需要延迟那么将该事件发送到MQ 队列中,为需要延迟的消息设置一个TTL;
  2. TTL到期后就会自动进入设置好的DLX,然后由DLX转发到配置好的实际消费队列;
  3. 消费该队列的延迟消息,处理事件。

方案优劣:

优点:
大品牌组件,用的放心。如果面临大数据量需求可以很容易的横向扩展,同时消息支持持久化,有问题可回滚。
缺点:

  1. 配置麻烦,额外增加一个死信交换机和一个死信队列的配置;
  2. RabbitMQ 是一个消息中间件,TTL 和 DLX 只是他的一个特性,将延迟队列绑定在一个功能软件的某一个特性上,可能会有风险。不要杠,当你们组不用 RabbitMQ 的时候迁移很痛苦;
  3. 消息队列具有先进先出的特点,如果第一个进入队列的消息 A 的延迟是10分钟,第二个进入队列的消息B 的延迟是5分钟,期望的是谁先到 TTL谁先出,但是事实是B已经到期了,而还要等到 A 的延迟10分钟结束A先出之后,B 才能出。所以在设计的时候需要考虑不同延迟的消息要放到不同的队列。另外该问题官方已经给出了插件来支持:插件地址。

延迟队列的实现有两种方式:

通过消息过期后进入死信交换器,再由交换器转发到延迟消费队列,实现延迟功能;
使用 RabbitMQ-delayed-message-exchange 插件实现延迟功能。
针对任务丢失的代价过大,高并发的场景
优点: 支持集群,分布式,高并发场景;
缺点: 引入额外的消息队列,增加项目的部署和维护的复杂度。
场景:为一个委托指定期限,委托到期后,委托关系终止,相关业务权限移交回原拥有者 这里采用的是RabbitMq的死信队列加TTL消息转化为延迟队列的方式(RabbitMq没有延时队列)

①声明一个队列设定其的死信队列  
@Configuration
public class MqConfig {public static final String GLOBAL_RABBIT_TEMPLATE = "rabbitTemplateGlobal";public static final String DLX_EXCHANGE_NAME = "dlxExchange";public static final String AUTH_EXCHANGE_NAME = "authExchange";public static final String DLX_QUEUE_NAME = "dlxQueue";public static final String AUTH_QUEUE_NAME = "authQueue";public static final String DLX_AUTH_QUEUE_NAME = "dlxAuthQueue";@Bean@Qualifier(GLOBAL_RABBIT_TEMPLATE)public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);return rabbitTemplate;}@Bean@Qualifier(AUTH_EXCHANGE_NAME)public Exchange authExchange() {return ExchangeBuilder.directExchange (AUTH_EXCHANGE_NAME).durable (true).build ();}/*** 死信交换机* @return*/@Bean@Qualifier(DLX_EXCHANGE_NAME)public Exchange dlxExchange() {return ExchangeBuilder.directExchange (DLX_EXCHANGE_NAME).durable (true).build ();}/*** 记录日志的死信队列* @return*/@Bean@Qualifier(DLX_QUEUE_NAME)public Queue dlxQueue() {// Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)return QueueBuilder.durable (DLX_QUEUE_NAME).build ();}/*** 委托授权专用队列* @return*/@Bean@Qualifier(AUTH_QUEUE_NAME)public Queue authQueue() {return QueueBuilder.durable (AUTH_QUEUE_NAME).withArgument("x-dead-letter-exchange", DLX_EXCHANGE_NAME).withArgument("x-dead-letter-routing-key", "dlx_auth").build ();}/*** 委托授权专用死信队列* @return*/@Bean@Qualifier(DLX_AUTH_QUEUE_NAME)public Queue dlxAuthQueue() {// Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)return QueueBuilder.durable (DLX_AUTH_QUEUE_NAME).withArgument("x-dead-letter-exchange", DLX_EXCHANGE_NAME).withArgument("x-dead-letter-routing-key", "dlx_key").build ();}@Beanpublic Binding bindDlxQueueExchange(@Qualifier(DLX_QUEUE_NAME) Queue dlxQueue, @Qualifier(DLX_EXCHANGE_NAME) Exchange dlxExchange){return BindingBuilder.bind (dlxQueue).to (dlxExchange).with ("dlx_key").noargs ();}/*** 委托授权专用死信队列绑定关系* @param dlxAuthQueue* @param dlxExchange* @return*/@Beanpublic Binding bindDlxAuthQueueExchange(@Qualifier(DLX_AUTH_QUEUE_NAME) Queue dlxAuthQueue, @Qualifier(DLX_EXCHANGE_NAME) Exchange dlxExchange){return BindingBuilder.bind (dlxAuthQueue).to (dlxExchange).with ("dlx_auth").noargs ();}/*** 委托授权专用队列绑定关系* @param authQueue* @param authExchange* @return*/@Beanpublic Binding bindAuthQueueExchange(@Qualifier(AUTH_QUEUE_NAME) Queue authQueue, @Qualifier(AUTH_EXCHANGE_NAME) Exchange authExchange){return BindingBuilder.bind (authQueue).to (authExchange).with ("auth").noargs ();}}②发送含过期时间的消息  向授权交换机,发送路由为"auth"的消息(指定了业务所需的超时时间) =》发向MqConfig.AUTH_QUEUE_NAME 队列  
rabbitTemplate.convertAndSend(MqConfig.AUTH_EXCHANGE_NAME, "auth", "类型:END,信息:{id:1,fromUserId:111,toUserId:222,beginData:20201204,endData:20211104}", message -> {/*** MessagePostProcessor:消息后置处理* 为消息设置属性,然后返回消息,相当于包装消息的类*///业务逻辑:过期时间=xxxxString ttl = "5000";//设置消息的过期时间message.getMessageProperties ().setExpiration (ttl);return message;});
复制代码
③超时后队列MqConfig.AUTH_QUEUE_NAME会将消息转发至其配置的死信路由"dlx_auth",监听该死信队列即可消费定时的消息/*** 授权定时处理* @param channel* @param message*/@RabbitListener(queues = MqConfig.DLX_AUTH_QUEUE_NAME)public void dlxAuthQ(Channel channel, Message message) throws IOException {System.out.println ("\n死信原因:" + message.getMessageProperties ().getHeaders ().get ("x-first-death-reason"));//1.判断消息类型:1.BEGIN 2.ENDtry {//2.1 类型为授权到期(END)//2.1.1 修改报件办理人//2.1.2 修改授权状态为0(失效)//2.2 类型为授权开启(BEGIN)//2.2.1 修改授权状态为1(开启)System.out.println (new String(message.getBody (), Charset.forName ("utf8")));channel.basicAck (message.getMessageProperties ().getDeliveryTag (),  false);System.out.println ("已处理,授权相关信息修改成功");} catch (Exception e) {//拒签消息channel.basicNack (message.getMessageProperties ().getDeliveryTag (), false, false);System.out.println ("授权相关信息处理失败, 进入死信队列记录日志");}}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://xiahunao.cn/news/2806232.html

如若内容造成侵权/违法违规/事实不符,请联系瞎胡闹网进行投诉反馈,一经查实,立即删除!

相关文章

RuntimeError: CUDNN_STATUS_EXECUTION_FAILED

问题描述&#xff1a; 运行代码时候报错&#xff1a; 原因&#xff1a;pytorch与cuda版本不对&#xff0c;需要重新安装。不过我在复现代码的时候一般是要求特定的环境&#xff0c;不然会有其他错误&#xff0c;所以选择其他解决办法。 解决方案&#xff1a; 在train.py开头…

跨界计算与控制,强化显控和UI, 君正MPU再添新旗舰--Ingenic MPU X2600隆重发布

近日&#xff0c;北京君正隆重发布MPU芯片新产品X2600。该产品以商业和工业应用的数个细分领域为重点目标市场&#xff0c;兼顾通用处理器应用需求。无论从CPU结构的设计&#xff0c;还是专门控制器和接口的配备&#xff0c;都体现了北京君正MPU团队“技术路线上追求自主跨界&a…

鸿蒙开发之Profiler性能分析

一、Profiler性能分析器简介 应用或服务的性能较差时,可能表现为响应速度慢、动画播放不流畅、卡顿、崩溃或极其耗电。为了避免出现这些性能问题,需要通过一系列性能分析工具来确定应用或服务对哪方面资源(例如CPU、内存、显卡、网络和设备电池)的使用率比较高。DevEco St…

代码随想录算法训练营day25|216.组合总和III

216.组合总和III 题目链接/文章讲解&#xff1a;代码随想录 视频讲解&#xff1a;和组合问题有啥区别&#xff1f;回溯算法如何剪枝&#xff1f;| LeetCode&#xff1a;216.组合总和III_哔哩哔哩_bilibili 跟77题差不多&#xff0c;要搞清楚k确定了递归的深度 依旧用回溯三部…

Facebook的数字社交使命:连接世界的下一步

在数字化时代&#xff0c;社交媒体已成为人们生活的重要组成部分&#xff0c;而Facebook作为其中最具影响力的平台之一&#xff0c;一直以来都在努力履行着自己的使命——连接世界。然而&#xff0c;随着时代的变迁和技术的发展&#xff0c;Facebook正在不断探索着连接世界的下…

【Logback】Logback 日志框架的架构

目录 1、Logger&#xff08;记录器&#xff09; &#xff08;1&#xff09;有效级别和级别继承 &#xff08;2&#xff09;日志打印和日志筛选 &#xff08;3&#xff09;记录器命名 2、Appenders&#xff08;追加器&#xff09; 3、Layouts&#xff08;布局&#xff09;…

提示工程(Prompt Engineering)、微调(Fine-tuning) 和 嵌入(Embedding)

主要参考资料&#xff1a; 还没搞懂嵌入&#xff08;Embedding&#xff09;、微调&#xff08;Fine-tuning&#xff09;和提示工程&#xff08;Prompt Engineering&#xff09;&#xff1f;: https://blog.csdn.net/DynmicResource/article/details/133638079 B站Up主Nenly同学…

智能高压森林应急消防泵|保障森林安全|深圳恒峰

随着科技的不断发展&#xff0c;我们的生活质量得到了显著提升。在森林保护领域&#xff0c;一项创新技术正在发挥着关键作用——智能高压森林应急消防泵。这种设备不仅提高了灭火效率&#xff0c;更为森林资源的安全保驾护航。 在过去&#xff0c;面对森林火灾&#xff0c;消防…

学习Python分支结构不走弯路

1.单分支语句 """ 语法&#xff1a; if 表达式:执行语句 执行流程&#xff1a;当表达式成立的时候&#xff0c;执行语句&#xff0c;否则不执行 """age int(input(请输入你的年龄&#xff1a;)) if age > 18:print(欢迎光临&#xff01;) …

PyTorch基础:Tensor类型张量的构建与相互转换

PyTorch基础&#xff1a;Tensor类型张量的构建与相互转换 &#x1f308; 个人主页&#xff1a;高斯小哥 &#x1f525; 高质量专栏&#xff1a;Matplotlib之旅&#xff1a;零基础精通数据可视化、Python基础【高质量合集】、PyTorch零基础入门教程 &#x1f448; 希望得到您的订…

【Java】Java基础(实验一)

目录 一、实验目的 二、实验内容 三、实验小结 一、实验目的 掌握Java程序的编辑、调试与运行&#xff1b;了解Java引用类型&#xff0c;掌握数组的定义和引用。掌握Java基本数据类型和输入输出。掌握Java程序结构 二、实验内容 1.JDK的环境变量设置及测试。 &#xff08…

2023年12月CCF-GESP编程能力等级认证C++编程六级真题解析

一、单选题(共15题,共30分) 第1题 关于C++类和对象的说法,错误的是( )。 A:在C++中,一切皆对象,即便是字面量如整数5等也是对象 B:在C++中,可以自定义新的类,并实例化为新的对象 C:在C++中,内置函数和自定义函数,都是类或者对象 D:在C++中,可以在自定义函数中…

fly-barrage 前端弹幕库(1):项目介绍

fly-barrage 是我写的一个前端弹幕库&#xff0c;由于经常在 Bilibili 上看视频&#xff0c;所以对网页的弹幕功能一直蛮感兴趣的&#xff0c;所以做了这个库&#xff0c;可以帮助前端快速的实现弹幕功能。 项目官网地址&#xff1a;https://fly-barrage.netlify.app/&#xff…

c++获取本地所有IP地址,以及域名解析

#include <iostream> using namespace std; #define _WINSOCK_DEPRECATED_NO_WARNINGS #include <WinSock2.h> #pragma comment(lib,"WS2_32.lib")class CInitSock { public:CInitSock(){//必须要注册网络库WSADATA wsd;if (::WSAStartup(MAKEWORD(2, 2)…

【k8s资源调度-Deployment】

1、标签和选择器 1.1 标签Label 配置文件&#xff1a;在各类资源的sepc.metadata.label 中进行配置通过kubectl 命令行创建修改标签&#xff0c;语法如下 创建临时label&#xff1a;kubectl label po <资源名称> apphello -n <命令空间&#xff08;可不加&#xff0…

【SpringBoot】Spring常用注解总结

目录 ⭐spring springmvc和springboot的区别 Autowired 和Resource的区别和联系 1. SpringBootApplication 2. Spring Bean 相关 2.1. Autowired 2.2. Component,Repository,Service, Controller 2.3. RestController 2.4. Scope 2.5. Configuration 3. 处理常见的 HT…

CentOS和Ubuntu之间的区别和联系

CentOS&#xff08;Community ENTerprise Operating System&#xff09;和Ubuntu是两种流行的Linux发行版&#xff0c;它们在企业和个人用户中都有广泛的应用。尽管它们都是基于Linux内核&#xff0c;但它们在设计理念、更新策略、包管理系统等方面存在一些关键的区别和联系。下…

力扣 724. 寻找数组的中心下标

思路&#xff1a; 创建两个变量sum和sum1&#xff0c;sum代表左边元素的和&#xff0c;sum1代表右边元素的和 然后假设从数组下标0开始&#xff0c;一直到最后一个作为中心下标 如果sumsum1&#xff0c;返回此时的中心下标 如果所有下标循环完了&#xff0c;发现没有return…

Sqli-labs靶场第8关详解[Sqli-labs-less-8]

Sqli-labs-Less-8 前言&#xff1a; SQL注入的三个条件&#xff1a; ①参数可控&#xff1b;&#xff08;从参数输入就知道参数可控&#xff09; ②参数过滤不彻底导致恶意代码被执行&#xff1b;&#xff08;需要在测试过程中判断&#xff09; ③参数带入数据库执行。&#…

Pytorch安装如何使用命令确认CUDA版本

Pytorch安装如何使用命令确认CUDA版本 一、NVIDIA版本确认命令解析二、Pytorch对应的NVIDIA版本选择 欢迎学习交流&#xff01; 邮箱&#xff1a; z…1…6.com 网站&#xff1a; https://zephyrhours.github.io/ 一、NVIDIA版本确认命令解析 在使用深度学习的Pytorch库时&…