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://www.rhkb.cn/news/263308.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

数据结构与算法相关题解20240225

数据结构与算法相关题解20240225

数据结构与算法相关题解20240225 一、58. 最后一个单词的长度二、48. 旋转图像三、69. x 的平方根四、50. Pow(x, n) 一、58. 最后一个单词的长度 简单 给你一个字符串 s&#xff0c;由若干单词组成&#xff0c;单词前后用一些空格字符隔开。返回字符串中 最后一个 单词的长度…
阅读更多...
自定义Chrome的浏览器开发者工具DevTools界面的字体和样式

自定义Chrome的浏览器开发者工具DevTools界面的字体和样式

Chrome浏览器开发者工具默认的字体太小&#xff0c;想要修改但没有相关设置。 外观——字体可以自定义字体&#xff0c;但大小不可以调整。 github上有人给出了方法 整理为中文教程&#xff1a; 1.打开浏览器开发者工具&#xff0c;点开设置——实验&#xff0c;勾上红框设…
阅读更多...
Excel工作表控件实现滚动按钮效果

Excel工作表控件实现滚动按钮效果

实例需求&#xff1a;工作表中有多个Button控件&#xff08;工作表Form控件&#xff09;和一个ScrollBar控件&#xff08;工作表ActiveX控件&#xff0c;名称为ScrollBar2&#xff09;&#xff0c;需要实现如下图所示效果。点击ScrollBar控件实现按钮的滚动效果&#xff0c;实际…
阅读更多...
群晖NAS DSM7.2.1安装宝塔之后无法登陆账号密码问题解决

群晖NAS DSM7.2.1安装宝塔之后无法登陆账号密码问题解决

宝塔的安装就不在这赘述了&#xff0c;只说下&#xff0c;启动之后默认账号密码无法登陆的问题。 按照上面给出的账号密码&#xff0c;无法登陆 然后点忘记密码&#xff0c;由于是docker安装的&#xff0c;根目录下没有/www/server/panel 。 也没有bt命令 要怎么修改呢。 既然…
阅读更多...
【Python笔记-设计模式】适配器模式

【Python笔记-设计模式】适配器模式

一、说明 适配器模式是一种结构型模式&#xff0c;它使接口不兼容的对象能够相互合作 (一) 解决问题 主要解决接口不兼容问题 (二) 使用场景 当系统需要使用现有的类&#xff0c;但类的接口不符合需求时当需要一个统一的输出接口&#xff0c;但输入类型不可预知时当需要创…
阅读更多...
Flink join详解（含两类API及coGroup、connect详解）

Flink join详解（含两类API及coGroup、connect详解）

Flink SQL支持对动态表进行复杂而灵活的连接操作。 为了处理不同的场景&#xff0c;需要多种查询语义&#xff0c;因此有几种不同类型的 Join。 默认情况下&#xff0c;joins 的顺序是没有优化的。表的 join 顺序是在 FROM 从句指定的。可以通过把更新频率最低的表放在第一个、…
阅读更多...
网络编程、UDP、TCP

网络编程、UDP、TCP

计算机网络 就是将地理位置不同的具有独立功能的多台计算及外部设备&#xff0c;通过通信线路连接起来&#xff0c;在网络操作系统、网络管理软件以及网络通信协议的管理和协调下&#xff0c;实现资源共享和信息传递的计算机系统 目的 传播交流信息、数据交换、通信 如何做…
阅读更多...
*MYSQL--索引--内部原理

*MYSQL--索引--内部原理

MYSQL的索引根据功能,主要有三大类型: 1.HASH索引 2.二叉树 3.BTREE索引 一:HASH索引 1.内部原理: 在设置了某列为索引列之后,并且开始或者将要在相应索引列创建数据的时候,系统通过某种算法 F(X) 自动计算出来一个十六进制的哈希值,这个哈希值能够对应相应的字段值 所以…
阅读更多...
链表和顺序表的优劣分析及其时间、空间复杂度分析

链表和顺序表的优劣分析及其时间、空间复杂度分析

链表和顺序表的优劣分析及其时间、空间复杂度分析 一、链表和顺序表的优劣分析二、算法复杂度<font face "楷体" size 5 color blue>//上面算法的执行次数大致为&#xff1a;F&#xff08;N&#xff09; N^22*N10;   N 10,F(10) 1002010 130次   N 1…
阅读更多...
大模型+影像：智能手机“上春山”

大模型+影像：智能手机“上春山”

这个春节假期&#xff0c;一首《上春山》火了。吃瓜群众热热闹闹学了一个假期的“春山学”&#xff0c;了解了抢占C位的各种技巧。 假期过去&#xff0c;开工大吉&#xff0c;手机行业开始抢占今年的C位。那么问题来了&#xff0c;今年智能手机最大的机会点在哪里&#xff1f;答…
阅读更多...
修复Microsoft Edge WebView2无法安装的问题

修复Microsoft Edge WebView2无法安装的问题

修复Microsoft Edge WebView2无法安装的问题 场景解决方案 场景 系统&#xff1a;win11 电脑&#xff1a;联想14 前提&#xff1a;使用Geek Uninstaller强制删除了Microsoft Edge WebView2 同时下载了clash verge。 发现根本无法运行&#xff08;点击了无任何反应且图标颜色…
阅读更多...
【Linux】 yum命令使用

【Linux】 yum命令使用

yum命令 yum&#xff08; Yellow dog Updater, Modified&#xff09; 是一个在 Fedora、CentOS 及其它一些基于 RPM 的 Linux 发行版中使用的包管理器。它允许用户自动安装、更新、配置和删除软件包。yum 由 Python 写成&#xff0c;基于 RPM&#xff08;Red Hat Package Mana…
阅读更多...
Sublime Text4配置C#运行环境

Sublime Text4配置C#运行环境

这里写自定义目录标题 前言部署.NET环境Sublime Text4配置C#编译环境1. 下载插件 运行测试 前言 今天把家里的9年前的远古神机搬了出来&#xff0c;重装了个win7的精简版&#xff0c;本打算装个VScode测试一下是否能写C#代码&#xff0c;结果是可以的&#xff0c;但&#xff0…
阅读更多...
【Docker】有用的命令

【Docker】有用的命令

文章目录 DockerDocker 镜像与容器的差异Docker的好处Hypervisor运维 一、安装docker二、启动docker三、获取docker镜像四、创建镜像使用命令行创建镜像使用dockerfile创建镜像 五、docker报错 Docker docker镜像&#xff08;Image&#xff09; docker镜像类似于虚拟机镜像&…
阅读更多...
MacOSX删除HBuilderX

MacOSX删除HBuilderX

1. 删除应用程序 如下图。MacOSX&#xff0c;打开访达&#xff0c;左侧点击应用程序&#xff0c;进入应用程序窗口&#xff0c;找到HBuilderX&#xff0c;右键菜单&#xff0c;点击移到废纸篓 2. 删除数据目录 打开操作系统终端&#xff0c;输入如下命令&#xff1a; open $H…
阅读更多...
性能优化——canvas 加载海量图

性能优化——canvas 加载海量图

背景 公司的在线设计稿平台的画板列表页开发时由于数据量不足&#xff0c;未能测出关于画板列表页性能问题&#xff0c;在经过用户一段时间的使用后出现了关于初始化卡顿、缩放卡顿等问题&#xff0c;画板列表页采用了vue-konva 原因 关于画板列表为何卡顿有如下几点原因 1、…
阅读更多...
uni-app 人脸识别 App端

uni-app 人脸识别 App端

文章目录 背景介绍开发前准备基础版获取视频流人脸识别版本这时候就可以开心的调试了背景介绍 本文介绍如何制作人脸打卡等类似功能的实现。 使用nvue+live-pusher来实现。在App端这是成本较低的可以控制样式的方案了 实现了两个版本 基础版本：视频流 => 抓拍照片 => 传…
阅读更多...
手写redux和applyMiddleware中间件react示例

手写redux和applyMiddleware中间件react示例

目录 一 核心代码 1.reducer 2.store.js 二 关于context API的使用 1. MyContext 2. createContext 3. ContextProvider 4. connect 三 组件验证效果 1. Todo 2. TodoList 3.TodoItem 4.TodoInput 5. App组件引入Todo组件 一 核心代码 1.reducer // 新增列表数…
阅读更多...
鸿蒙DevEco Service开发准备与使用

鸿蒙DevEco Service开发准备与使用

DevEco低代码是一个基于Serverless和ArkUI的端云一体化低代码开发平台&#xff0c;可通过拖拽式开发&#xff0c;可视化配置构建元服务。打通HarmonyOS云侧与端侧能力&#xff0c;轻松实现HMS Core和AGC Serverless能力的调用。通过与元服务生态、HMS Core、AGC Serverless平台…
阅读更多...
igolang学习2，golang开发配置国内镜像

igolang学习2，golang开发配置国内镜像

go env -w GO111MODULEon go env -w GOPROXYhttps://goproxy.cn,direct
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023