RabbitMQ 死信队列应用

2024-02-01 10:52
文章标签 应用 队列 rabbitmq 死信

本文主要是介绍RabbitMQ 死信队列应用，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1. 概念

死信队列（Dead Letter Queue）是在消息队列系统中的一种特殊队列，用于存储无法被消费的消息。消息可能会因为多种原因变成“死信”，例如消息过期、消息被拒绝、消息队列长度超过限制等。当消息变成“死信”时，它们会被路由到死信队列中，以便进行进一步处理或分析。 死信队列能够帮助系统进行消息跟踪、监控和处理异常情况，是消息队列系统中的重要组成部分。

2. 应用场景

死信队列在消息队列系统中有多种应用场景，包括但不限于以下几个方面：

  • 延迟消息处理：实现延迟消息投递，例如实现消息的定时投递、消息重试机制等。

  • 任务调度：用于实现任务调度系统，例如延迟执行任务、失败重试任务等。

  • 异常处理：处理消息消费失败或超时的情况，对异常消息进行统一处理。

  • 业务流程控制：实现业务流程中的状态控制和超时处理，例如订单超时取消、支付超时处理等。

  • 监控和统计：对异常消息进行统计和分析，用于系统性能监控和问题排查。

这些应用场景展示了死信队列的灵活性和实用性，在实际系统开发中具有广泛的应用价值。

3. 造成消息进入死信队列的原因

消息成为死信的原因有以下几种：

  • 消息被拒绝（basic.reject或basic.nack），并且requeue标志被设置为false。若参数requeue为true，则表示还可以将此跳消息重新塞回普通队列，若为false则消息被拒绝后直接进入死信队列。

  • 消息过期。在生产者设置生产时设置，若消费者未在过期时间内消费消息，则消息被转发到死信队列中。（"x-message-ttl"）

  • 队列达到最大长度。当普通队列中消息堆积数量长度达到了maxLength，则会将新接收的消息转发到死信队列中去，从而避免消息丢失。

4. 死信队列工作流程图

5. 代码示例

5.1 引入依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>2.7.15</version>
</dependency>

5.2 RabbitMQ配置

@Configuration
public class RabbitConfig {/*** 死信队列消息模型构建----------------------------------------------------------------------------------**/// 创建普通队列@Beanpublic Queue basicQueue() {Map<String, Object> params = new HashMap<>(8);// x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，params.put("x-dead-letter-exchange", Exchange.DEMO_DEAD_LETTER_EXCHANGE);// x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。params.put("x-dead-letter-routing-key", RoutingKey.DEMO_DEAD_ROUTING_KEY);// 注意这里是毫秒单位，这里我们给10秒params.put("x-message-ttl", 10*1000);return new Queue(MyQueue.DEMO_CONSUMER_QUEUE, true, false, false, params);}//创建“基本消息模型”的基本交换机，面向生产者@Beanpublic TopicExchange basicExchange() {//创建并返回基本交换机实例return new TopicExchange(Exchange.DEMO_BASIC_NORMAL_EXCHANGE, true, false);}//创建“基本消息模型”的基本绑定（基本交换机+基本路由），面向生产者@Beanpublic Binding basicBinding() {//创建并返回基本消息模型中的基本绑定(注意这里是正常交换机跟死信队列绑定在一定，不叫死信路由)return BindingBuilder.bind(basicQueue()).to(basicExchange()).with(RoutingKey.DEMO_ROUTING_KEY);}// 创建死信交换机@Beanpublic TopicExchange deadLetterExchange() {//创建并返回死信交换机实例return new TopicExchange(Exchange.DEMO_DEAD_LETTER_EXCHANGE, true, false);}// 创建第二个中转站// 创建死信队列@Beanpublic Queue deadLetterQueue() {return new Queue(MyQueue.DEMO_DEAD_LETTER_QUEUE, true);}// 创建死信路由及其绑定@Beanpublic Binding deadLetterBinding() {return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(RoutingKey.DEMO_DEAD_ROUTING_KEY);}public static class Exchange {public static final String DEMO_BASIC_NORMAL_EXCHANGE = "demo.basic.exchange";public static final String DEMO_DEAD_LETTER_EXCHANGE = "demo.dead.letter.exchange";}public static class RoutingKey {//交换机与报表队列绑定的RoutingKeypublic static final String DEMO_ROUTING_KEY = "demo.basic.routing.key";public static final String DEMO_DEAD_ROUTING_KEY = "demo.dead.routing.key";}/*** 队列名称* @author peng.zhang* @date 2024/01/30*/public static class MyQueue {//报表队列名称public static final String DEMO_CONSUMER_QUEUE = "demo.basic.queue";//死信队列名称public static final String DEMO_DEAD_LETTER_QUEUE = "demo.dead.letter.queue";}
}

5.3 消息生产者

@RestController
@RequestMapping("/test")
@Slf4j
public class TestController {@Resourceprivate RabbitTemplate rabbitTemplate;/*** 发送消息到死信队列*/@PostMapping("/testDeadQueue")public String testDeadQueue() {// 设置生产者到交换机的确认回调rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {log.info("correlationData:{}, ack:{}, cause:{}", JSON.toJSONString(correlationData), ack, cause);});// 设置消息未被队列接收时的返回回调rabbitTemplate.setReturnCallback((message, replyCode, replyText, ex, routing) -> {log.info("message:{}, replyCode:{}, replyText:{}, exchange:{}, routingKey:{}", JSON.toJSONString(message),replyCode, replyText, ex, routing);});// 生成关联数据并发送消息到交换机CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());// 消息内容String messageBody = StrUtil.format("this message send at {}", DateUtil.format(LocalDateTime.now(), "yyyy-MM-dd HH:mm:ss"));rabbitTemplate.convertAndSend(RabbitConfig.Exchange.DEMO_BASIC_NORMAL_EXCHANGE, RabbitConfig.RoutingKey.DEMO_ROUTING_KEY, messageBody, correlationData);log.info(">>>>>{}, 发送消息:{}", DateUtil.format(LocalDateTime.now(), "yyyy-MM-dd HH:mm:ss"), messageBody);return "OK";}}

5.4 消息消费者

@Component
@Slf4j
public class DeadLetterConsumer {/*** 监听 DEMO_CONSUMER_QUEUE 并处理传入的消息。* 为测试目的抛出 IOException 以模拟异常。** @param messageBody 消息负载* @param headers     消息头* @param channel     用于消息确认的通道* @throws IOException 如果抛出异常*/@RabbitListener(queues = RabbitConfig.MyQueue.DEMO_CONSUMER_QUEUE)@RabbitHandlerpublic void testBasicQueueAndThrowsException(@Payload String messageBody, @Headers Map<String, Object> headers, Channel channel) throws IOException {/*** Delivery Tag 用来标识信道中投递的消息。RabbitMQ 推送消息给 Consumer 时，会附带一个 Delivery Tag，* 以便 Consumer 可以在消息确认时告诉 RabbitMQ 到底是哪条消息被确认了。* RabbitMQ 保证在每个信道中，每条消息的 Delivery Tag 从 1 开始递增。*/Long tag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);log.info(">>>>>{} 普通队列消费, tag = {}, 消息内容:{}", DateUtil.format(LocalDateTime.now(), "yyyy-MM-dd HH:mm:ss"), tag, messageBody);/***  multiple 取值为 false 时，表示通知 RabbitMQ 当前消息被确认*  如果为 true，则额外将比第一个参数指定的 delivery tag 小的消息一并确认*/// ACK,确认一条消息已经被消费
//        channel.basicAck(deliveryTag, false);// 对应的业务操作。。。。。// doBusiness();// 模拟消息拒绝channel.basicNack(tag, false, false);}/*** 处理业务逻辑*/private void doBusiness() {System.out.println("here do some business code");}/*** 监听死信队列并处理消息。** @param data    消息内容* @param tag     消息标签* @param channel 通道*/@RabbitListener(queues = RabbitConfig.MyQueue.DEMO_DEAD_LETTER_QUEUE)@RabbitHandlerpublic void fromDeadLetter(@Payload String data, @Header(AmqpHeaders.DELIVERY_TAG) long tag, Channel channel) {log.info(">>>>>{} 死信队列消费, tag = {}, 消息内容:{}", DateUtil.format(LocalDateTime.now(), "yyyy-MM-dd HH:mm:ss"), tag, data);// 对应的业务操作。。。。。}
}

5.5 YML配置

spring:rabbitmq:username: rabbitmqpassword: rabbitmqport: 5672host: 127.0.0.1#publisher-confirm-type参数有三个可选值：#SIMPLE：会触发回调方法，相当于单个确认（发一条确认一条）。#CORRELATED：消息从生产者发送到交换机后触发回调方法。#NONE（默认）：关闭发布确认模式。publisher-confirm-type: correlatedtemplate:receive-timeout: 1800000reply-timeout: 1800000retry:enabled: falselistener:direct:retry:enabled: truedefault-requeue-rejected: falsesimple:retry:# 是否开启消费者重试（为false时关闭消费者重试，这时消费端代码异常会一直重复收到消息）enabled: true# 最大重试次数max-attempts: 1# 重试间隔时间（单位毫秒）initial-interval: 10000# 重试最大时间间隔（单位毫秒）max-interval: 300000# 应用于前一重试间隔的乘法器multiplier: 5default-requeue-rejected: false

5.6 控制台输出

从控制台可以看出，消息被拒绝后，大概10秒后死信队列消息被消费。

这篇关于RabbitMQ 死信队列应用的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/667042。

相关文章

中文分词jieba库的使用与实景应用（一）

中文分词jieba库的使用与实景应用（一）

知识星球：https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义： 精确模式（默认模式）： 全模式： 搜索引擎模式： paddle 模式（基于深度学习的分词模式）： 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取
阅读更多...
水位雨量在线监测系统概述及应用介绍

水位雨量在线监测系统概述及应用介绍

在当今社会，随着科技的飞速发展，各种智能监测系统已成为保障公共安全、促进资源管理和环境保护的重要工具。其中，水位雨量在线监测系统作为自然灾害预警、水资源管理及水利工程运行的关键技术，其重要性不言而喻。 一、水位雨量在线监测系统的基本原理 水位雨量在线监测系统主要由数据采集单元、数据传输网络、数据处理中心及用户终端四大部分构成，形成了一个完整的闭环系统。 数据采集单元：这是系统的“眼睛”，
阅读更多...
hdu1180（广搜+优先队列）

hdu1180（广搜+优先队列）

此题要求最少到达目标点T的最短时间，所以我选择了广度优先搜索，并且要用到优先队列。 另外此题注意点较多，比如说可以在某个点停留，我wa了好多两次，就是因为忽略了这一点，然后参考了大神的思想，然后经过反复修改才AC的 这是我的代码 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<
阅读更多...
csu 1446 Problem J Modified LCS （扩展欧几里得算法的简单应用）

csu 1446 Problem J Modified LCS （扩展欧几里得算法的简单应用）

这是一道扩展欧几里得算法的简单应用题，这题是在湖南多校训练赛中队友ac的一道题，在比赛之后请教了队友，然后自己把它a掉 这也是自己独自做扩展欧几里得算法的题目 题意：把题意转变下就变成了：求d1*x - d2*y = f2 - f1的解,很明显用exgcd来解 下面介绍一下exgcd的一些知识点：求ax + by = c的解 一、首先求ax + by = gcd(a,b)的解 这个
阅读更多...
hdu1394(线段树点更新的应用)

hdu1394(线段树点更新的应用)

题意：求一个序列经过一定的操作得到的序列的最小逆序数 这题会用到逆序数的一个性质，在0到n-1这些数字组成的乱序排列，将第一个数字A移到最后一位，得到的逆序数为res-a+(n-a-1) 知道上面的知识点后，可以用暴力来解 代码如下： #include<iostream>#include<algorithm>#include<cstring>#include<stack>#in
阅读更多...
zoj3820(树的直径的应用)

zoj3820(树的直径的应用)

题意：在一颗树上找两个点，使得所有点到选择与其更近的一个点的距离的最大值最小。 思路：如果是选择一个点的话，那么点就是直径的中点。现在考虑两个点的情况，先求树的直径，再把直径最中间的边去掉，再求剩下的两个子树中直径的中点。 代码如下： #include <stdio.h>#include <string.h>#include <algorithm>#include <map>#
阅读更多...
【区块链 + 人才服务】可信教育区块链治理系统 | FISCO BCOS应用案例

【区块链 + 人才服务】可信教育区块链治理系统 | FISCO BCOS应用案例

伴随着区块链技术的不断完善，其在教育信息化中的应用也在持续发展。利用区块链数据共识、不可篡改的特性， 将与教育相关的数据要素在区块链上进行存证确权，在确保数据可信的前提下，促进教育的公平、透明、开放，为教育教学质量提升赋能，实现教育数据的安全共享、高等教育体系的智慧治理。 可信教育区块链治理系统的顶层治理架构由教育部、高校、企业、学生等多方角色共同参与建设、维护，支撑教育资源共享、教学质量评估、
阅读更多...
AI行业应用(不定期更新)

AI行业应用(不定期更新)

ChatPDF 可以让你上传一个 PDF 文件，然后针对这个 PDF 进行小结和提问。你可以把各种各样你要研究的分析报告交给它，快速获取到想要知道的信息。https://www.chatpdf.com/
阅读更多...
poj 3190 优先队列+贪心

poj 3190 优先队列+贪心

题意： 有n头牛，分别给他们挤奶的时间。 然后每头牛挤奶的时候都要在一个stall里面，并且每个stall每次只能占用一头牛。 问最少需要多少个stall，并输出每头牛所在的stall。 e.g 样例： INPUT: 51 102 43 65 84 7 OUTPUT: 412324 HINT: Explanation of the s
阅读更多...
poj 2431 poj 3253 优先队列的运用

poj 2431 poj 3253 优先队列的运用

poj 2431： 题意： 一条路起点为0， 终点为l。 卡车初始时在0点，并且有p升油，假设油箱无限大。 给n个加油站，每个加油站距离终点 l 距离为 x[i]，可以加的油量为fuel[i]。 问最少加几次油可以到达终点，若不能到达，输出-1。 解析： 《挑战程序设计竞赛》： “在卡车开往终点的途中，只有在加油站才可以加油。但是，如果认为“在到达加油站i时，就获得了一
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2