当前位置: 首页 > news >正文

RabbitMQ 高级特性之死信队列

1. 简介

在前面的高级特性中,我们介绍了重试机制和 TTL,那么产生下列问题:

  • 在重试机制中,当消费者消费消息发生异常时,会触发消息重发机制,由于我们配置了最大的重发次数,那么当超过这个次数后,若消息依然没有被成功消费,就需要将消息进行保存,等待下一次消费,那么,这条消息应该保存到哪里去呢?
  • 在 TTL 机制中,我们为队列和消息设置了过期时间,当超过这个时间后消息就会被删除,但是这条消息是需要被消费的,于是就需要将过期的消息保存下来,等待下次消费。但是,消息应该保存到哪里去呢?

在 RabbitMQ 中,将类似于结果的消息称为死信,那么就涉及到,变成死信的消息应该存储到哪里去?

2. 会产生死信的场景

  • 消息重发后,次数到达指定重发次数依然未被消费,就会成为死信
  • 消息到达过期时间依然没有被消费,就会成为死信
  • 队列已经满了,却依然由消息入队列,就会产生溢出,溢出的这部分消息就会成为死信

3. 死信队列

在 RabbitMQ 中,可以声明一个队列,这个队列专门用来存放死信,于是就成为死信队列。

死信队列的工作流程如下:

  1. 首先需要声明一个死信交换机,与普通队列进行绑定;
  2. 其次需要声明一个死信队列,与死信交换机进行绑定;
  3. 当普通队列中的消息成为死信后,就会被发送给死信交换机,然后由死信交换机分配给与之绑定的死信队列;
  4. 存储在死信队列中的死信会等待被别的消费者再次消费。

4. 配置死信交换机与死信队列

声明一个正常交换机,正常队列,死信交换机,死信队列,并将正常交换机与正常队列进行绑定,将正常队列与死信交换机进行绑定,将死信交换机与死信队列进行绑定,代码如下:

@Configuration
public class DLConfig {/*** 正常* @return*/@Bean("norQueue")public Queue norQueue() {return QueueBuilder.durable(Constants.NOR_QUEUE).ttl(10000) //过期时间 10s.deadLetterExchange(Constants.DL_EXCHANGE) //绑定死信交换机.deadLetterRoutingKey(Constants.DL_ROUTINGKEY).maxLength(10L) //队列长度为 10.build();}@Bean("norExchange")public DirectExchange norExchange() {return ExchangeBuilder.directExchange(Constants.NOR_EXCHANGE).build();}@Bean("norBind")public Binding norBind(@Qualifier("norExchange") DirectExchange directExchange,@Qualifier("norQueue") Queue queue) {return BindingBuilder.bind(queue).to(directExchange).with(Constants.NOR_ROUTINGKEY);}/*** 死信*/@Bean("dlQueue")public Queue dlQueue() {return QueueBuilder.durable(Constants.DL_QUEUE).build();}@Bean("dlExchange")public DirectExchange dlExchange() {return ExchangeBuilder.directExchange(Constants.DL_EXCHANGE).build();}@Bean("dlBind")public Binding dlBind(@Qualifier("dlExchange") DirectExchange directExchange,@Qualifier("dlQueue") Queue queue) {return BindingBuilder.bind(queue).to(directExchange).with(Constants.DL_ROUTINGKEY);}
}

在上面的代码中,我们指定了下面的条件:

  • 队列的过期时间为 10s
  • 队列长度为 10 

重试机制的配置如下:

spring:rabbitmq:listener:simple:acknowledge-mode: autoretry:enabled: true # 开启消费者失败重试initial-interval: 5000ms # 初始失败等待时⻓为5秒max-attempts: 5 # 最⼤重试次数(包括自身消费的⼀次)

在上面的配置中,设置的最大的重发次数为 5 次。

5. 验证消息重试超过规定次数是否进入死信队列

生产者代码如下:

    @RequestMapping("/dl")public String dl() {String messageInfo = "dl... ";rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo);return "消息发送成功";}

消费者代码如下:

@Component
@Slf4j
public class DLListener {@RabbitListener(queues = Constants.NOR_QUEUE)public void listener(Message message) throws IOException {String messageInfo = new String(message.getBody());long deliveryTag = message.getMessageProperties().getDeliveryTag();log.info("接收消息, message: {}, deliveryTag: {}", messageInfo, deliveryTag);int num = 1 / 0;log.info("消息消费完成");}
}

代码运行结果如下:

 

消息重发五次后抛出异常,观察 RabbitMQ 客户端后,这条消息已经存放进了死信队列中:

  

6. 验证 TTL 与死信队列

生产者代码如下:

    @RequestMapping("/dl")public String dl() {String messageInfo = "dl... ";rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo);return "消息发送成功";}

将正常队列的 TTL 设置为 10s,运行代码后,观察 RabbitMQ 客户端:

10s 前:

10s 后:

 

7. 验证队列溢出与死信队列

将正常队列的长度设置为 10,生产者向 RabbitMQ 发送 20 条数据,生产者代码如下:

    @RequestMapping("/dl")public String dl() {for (int i = 0; i < 20; i++) {String messageInfo = "dl... " + i;rabbitTemplate.convertAndSend(Constants.NOR_EXCHANGE, Constants.NOR_ROUTINGKEY, messageInfo);}return "消息发送成功";}

运行代码,观察 RabbitMQ 客户端:

可以看到,溢出的 10 条消息进入了死信队列。 

http://www.dtcms.com/a/266839.html

相关文章:

  • VS CodeC51 单片机开发环境搭建
  • 猫头虎 AI工具分享:一个网页抓取、结构化数据提取、网页爬取、浏览器自动化操作工具:Hyperbrowser MCP
  • Web前端工程化
  • 网安系列【4】之OWASP与OWASP Top 10:Web安全入门指南
  • 一探 3D 互动展厅的神奇构造​
  • Querybook:一个开源大数据查询分析工具
  • Workflow or 自主智能体?网易CoreAgent如何打造企业级智能体平台新范式
  • OpenSearch添加仪表盘(elastic、es)
  • 全面分析软考《系统分析师》和《系统架构设计师》论文差异
  • go基础语法10问(2)
  • MySQL 事务详解:从基础操作到隔离级别与 MVCC 原理
  • vue3引入海康监控视频组件并实现非分屏需求一个页面同时预览多个监控视频;
  • 本地部署项目文档管理网站 MkDocs 并实现外部访问
  • Centos安装Jenkins
  • ZigBee通信技术全解析:从协议栈到底层实现,全方位解读物联网核心无线技术
  • OpenCV 图像操作:颜色识别、替换与水印添加
  • 传统架构开发VS PREEvision:一场效率与可靠性的降维打击
  • [C/C++内存安全]_[中级]_[如何避免数组访问越界]
  • 【精华】QPS限流等场景,Redis其他数据结构优劣势对比
  • 7.4_面试_JAVA_
  • python学习打卡:DAY 18 推断聚类后簇的类型
  • 在 Vue 3 中全局使用 Suspense 组件
  • 【内存】Linux 内核优化实战 - kernel.numa_balancing
  • [Linux]内核态与用户态详解
  • 1.1_3_2 三种交换方式的性能分析
  • PHP从字符串到数值的类型转换
  • 后端密码加密:守护用户数据的钢铁长城
  • 第三章 基于rtthread标准库的串口和shell应用
  • vue 循环无限滚动表格
  • 用distance_transform 检测线性凸包