【初始RabbitMQ】死信队列的实现

死信的概念

死信,顾名思义就是无法被消费的消息,字面意思可以这样理解,一般来说,producer 将消息投递到 broker 或者直接到 queue 里了,consumer 从 queue 取出消息 进行消费,但某些时候由于特定的原因导致 queue 中的某些消息无法被消费,这样的消息如果没有 后续的处理,就变成了死信,有死信自然就有了死信队列

应用场景:为了保证订单业务的消息数据不丢失,需要使用到 RabbitMQ 的死信队列机制,当消息消费发生异常时,将消息投入死信队列中。还有比如说:用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

死信的来源

死信的来源主要是有以下几个部分:

  1. 消息 TTL 过期
  2. 队列达到最大长度(队列满了,无法再添加数据到 mq 中)
  3. 消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false

死信的实战

代码架构图

消息TTL过期

生产者代码:

public class Producer {private static final String NORMAL_EXCHANGE = "normal_exchange";public static void main(String[] args) throws IOException {Channel channel = RabbitMqUtils.getChannel();channel.exchangeDeclare(NORMAL_EXCHANGE,"direct");//设置消息TTL的时间/*** 设置AMQP(高级消息队列协议)消息的属性* AMQP.BasicProperties: 这是AMQP协议中用于定义消息属性的类。* new AMQP.BasicProperties().builder(): 这里创建了一个新的AMQP.BasicProperties对象,并通过调用其builder()方法来开始构建该对象的属性。* .expiration("10000"): 在构建过程中,通过.expiration()方法设置了消息的expiration属性。expiration属性用于定义消息的TTL,即消息在队列中等待消费的时间。如果在这段时间内消息没有被消费,它将被自动丢弃。* 在这里,expiration的值被设置为"10000",这意味着消息的TTL是10,000毫秒,也就是10秒。* .build(): 最后,通过调用.build()方法来结束构建过程并返回完全配置的AMQP.BasicProperties对象。*/AMQP.BasicProperties properties =new AMQP.BasicProperties().builder().expiration("10000").build();for (int i = 1; i < 11; i++) {String message = "info"+i;/*** 发送一个消息* 1.发送到那个交换机* 2.路由的KEY值是哪个 本次是队列的名称* 3.其他参数信息* 4.发送消息的消息体*/channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",properties,message.getBytes(StandardCharsets.UTF_8));}}
}

消费者1的代码:

public class Consumer01 {//普通交换机名称private static final String NORMAL_EXCHANGE = "normal_exchange";//死信交换机名称private static final String DEAD_EXCHANGE = "dead_exchange";public static void main(String[] args) throws IOException {Channel channel = RabbitMqUtils.getChannel();//声明死信和普通交换机 类型为directchannel.exchangeDeclare(NORMAL_EXCHANGE,"direct");channel.exchangeDeclare(DEAD_EXCHANGE,"direct");//声明死信队列String deadQueue = "dead_queue";/**生成一个队列* 1.队列名称* 2.队列里面的信息是否持久化(磁盘)默认情况时在内存* 3.该队列是否只供一个消费者进行消费 是否消费共享 true是允许* 4.是否自动删除 最后一个消费者断开连接之后 该队列是否自动删除 true自动删除 false不自动删除* 5.其他参数 延迟消息等*/channel.queueDeclare(deadQueue,false,false,false,null);//死信队列绑定死信交换机与routingKeychannel.queueBind(deadQueue,DEAD_EXCHANGE,"lisi");//正常队列绑定死信队列信息Map<String, Object> params = new HashMap<>();//正常队列设置死信交换机 参数 key 是固定值params.put("x-dead-letter-exchange", DEAD_EXCHANGE);//正常队列设置死信 routing-key 参数 key 是固定值params.put("x-dead-letter-routing-key", "lisi");String normalQueue = "normal-queue";channel.queueDeclare(normalQueue, false, false, false, params);channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");System.out.println("C1等待接收消息......");DeliverCallback deliverCallback = (consumerTag,delivery)->{String message = new String(delivery.getBody(),"UTF-8");System.out.println("C1 接收到消息"+message);};/*** 消费者信息* 1.消费哪个队列* 2.消费成功之后是否要自动应答 true自动应答 false手动应答* 3.消费者成功才能更改消费的回调* 4.消费者取消消费回调*/channel.basicConsume(normalQueue,true,deliverCallback,consumerTag->{});}
}

此时启动生产者观看RabbitMQ的变化

消费者2的代码(用于消耗死信队列中的信息): 

public class Consumer02 {private static final String DEAD_EXCHANGE = "dead_exchange";public static void main(String[] args) throws IOException {Channel channel = RabbitMqUtils.getChannel();channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);String deadQueue = "dead-queue";channel.queueDeclare(deadQueue, false, false, false, null);channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");System.out.println("等待接收死信队列消息.....");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("Consumer02 接收死信队列的消息" + message);};channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {});}
}

运行结果: 

队列达到了最大长度

消息生产者代码去掉 TTL 属性:

public class Producer {private static final String NORMAL_EXCHANGE = "normal_exchange";public static void main(String[] argv) throws Exception {Channel channel = RabbitMqUtils.getChannel();channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);//该信息是用作演示队列个数限制for (int i = 1; i < 11; i++) {String message = "info" + i;channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", null, message.getBytes());System.out.println("生产者发送消息:" + message);}}
}

C1 消费者修改以下代码:

 C2 消费者代码不变(启动 C2 消费者),运行结果如下:

如果 报错请登录rabbitMQ管理平台将此队列删除:

 消息被拒

消息生产者代码同上生产者一致

C1 消费者的代码:

public class Consumer01 {//普通交换机名称private static final String NORMAL_EXCHANGE = "normal_exchange";//死信交换机名称private static final String DEAD_EXCHANGE = "dead_exchange";public static void main(String[] args) throws IOException {Channel channel = RabbitMqUtils.getChannel();//声明死信和普通交换机 类型为directchannel.exchangeDeclare(NORMAL_EXCHANGE,"direct");channel.exchangeDeclare(DEAD_EXCHANGE,"direct");//声明死信队列String deadQueue = "dead-queue";/**生成一个队列* 1.队列名称* 2.队列里面的信息是否持久化(磁盘)默认情况时在内存* 3.该队列是否只供一个消费者进行消费 是否消费共享 true是允许* 4.是否自动删除 最后一个消费者断开连接之后 该队列是否自动删除 true自动删除 false不自动删除* 5.其他参数 延迟消息等*/channel.queueDeclare(deadQueue,false,false,false,null);//死信队列绑定死信交换机与routingKeychannel.queueBind(deadQueue,DEAD_EXCHANGE,"lisi");//正常队列绑定死信队列信息Map<String, Object> params = new HashMap<>();//正常队列设置死信交换机 参数 key 是固定值params.put("x-dead-letter-exchange", DEAD_EXCHANGE);//正常队列设置死信 routing-key 参数 key 是固定值params.put("x-dead-letter-routing-key", "lisi");String normalQueue = "normal-queue";channel.queueDeclare(normalQueue, false, false, false, params);channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");System.out.println("C1等待接收消息......");DeliverCallback deliverCallback = (consumerTag,delivery)->{String message = new String(delivery.getBody(), "UTF-8");if(message.equals("info5")){System.out.println("Consumer01 接收到消息" + message + "并拒绝签收该消息");//requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);}else {System.out.println("Consumer01 接收到消息"+message);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}};/*** 消费者信息* 1.消费哪个队列* 2.消费成功之后是否要自动应答 true自动应答 false手动应答* 3.消费者成功才能更改消费的回调* 4.消费者取消消费回调*/channel.basicConsume(normalQueue,false,deliverCallback,consumerTag->{});}
}

C2 消费者代码不变

启动消费者 1 然后再启动消费者 2 ,运行结果如下:

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

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

相关文章

认识HarmonyOS

1.认识HarmonyOS 1.1.HarmonyOS简介 在中美贸易战的刺激下&#xff0c;国产操作系统HarmonyOS&#xff08;鸿蒙操作系统&#xff09;开始进入到大众的视野。 鸿蒙寓意为“万物起源”&#xff0c;发展至今已经经过了好几个迭代版本。 1.1.1.早期鸿蒙雏形 LiteOS 2015 年 5 月 …

STM32F103x 的时钟源

AHB (Advanced High-performance Bus) 高速总线&#xff0c;用来接高速外设的。 APB (Advanced Peripheral Bus) 低速总线&#xff0c;用来接低速外设的&#xff0c;包含APB1 和 APB2。 APB1&#xff1a;上面连接的是低速外设&#xff0c;包括电源接口、备份接口、 CAN 、 US…

【Linux网络】网络编程套接字(TCP)

目录 地址转换函数 字符串IP转整数IP 整数IP转字符串IP 关于inet_ntoa 简单的单执行流TCP网络程序 TCP socket API 详解及封装TCP socket 服务端创建套接字 服务端绑定 服务端监听 服务端获取连接 服务端处理请求 客户端创建套接字 客户端连接服务器 客户端…

QT-串口工具

一、演示效果 二、关键程序 &#xff1a; #include "mainwindow.h" #include "ui_mainwindow.h"#include <QMessageBox>MainWindow::MainWindow(QWidget *parent) :QMainWindow(parent),ui(new Ui::MainWindow),listPlugins(QList<TabPluginInt…

动态规划--持续更新篇

将数字变成0的操作次数 1.题目 2.思路 在numberOfSteps函数中&#xff0c;首先设置f[0]为0&#xff0c;因为0已经是0了&#xff0c;不需要任何步骤。然后&#xff0c;使用一个for循环从1迭代到输入的整数num。对于每个整数i&#xff0c;如果i是奇数&#xff0c;则将f[i]设置为…

静态时序分析:SDC约束命令set_driving_cell详解

相关阅读 静态时序分析https://blog.csdn.net/weixin_45791458/category_12567571.html?spm1001.2014.3001.5482 在上文中&#xff0c;我们不建议使用set_drive命令而是使用set_driving_cell命令&#xff0c;这是一个描述输入端口驱动能力更精确的方法。因为大多数情况下&…

SpringBoot实现缓存预热的几种常用方案

&#x1f3f7;️个人主页&#xff1a;牵着猫散步的鼠鼠 &#x1f3f7;️系列专栏&#xff1a;Java全栈-专栏 &#x1f3f7;️个人学习笔记&#xff0c;若有缺误&#xff0c;欢迎评论区指正 前些天发现了一个巨牛的人工智能学习网站&#xff0c;通俗易懂&#xff0c;风趣幽默&…

QEMU开发入门

1. 简介 QEMU&#xff08;Quick EMUlator&#xff09;是一个开源的虚拟化软件&#xff0c;它能够模拟多种硬件平台&#xff0c;并在这些平台上运行各种操作系统。QEMU可以在不同的主机架构之间进行虚拟化&#xff0c;例如x86、ARM、PowerPC、Risc-V等。QEMU是一个功能强大且灵…

LabVIEW开发FPGA的高速并行视觉检测系统

LabVIEW开发FPGA的高速并行视觉检测系统 随着智能制造的发展&#xff0c;视觉检测在生产线中扮演着越来越重要的角色&#xff0c;尤其是在质量控制方面。传统的基于PLC的视觉检测系统受限于处理速度和准确性&#xff0c;难以满足当前生产需求的高速和高精度要求。为此&#xf…

ACL权限、特殊位与隐藏属性的奥秘

1.2 操作步骤 # 1. 添加测试目录&#xff0c;用户&#xff0c;组&#xff0c;并将用户添加到组 ------------------- [rootlocalhost ~]# mkdir /project[rootlocalhost ~]# useradd zs[rootlocalhost ~]# useradd ls[rootlocalhost ~]# groupadd tgroup[rootlocalhost ~]# g…

软件提示找不到MSVCP140.dll是什么意思,修复MSVCP140.dll丢失的多个方法

msvcp140.dll 文件是 Microsoft Visual C 运行时库的一部分&#xff0c;具体来说它是 Visual Studio 2015 版本编译的C应用程序所依赖的一个动态链接库&#xff08;DLL&#xff09;文件。这个 DLL 文件包含了大量由Microsoft开发的标准C库函数&#xff0c;这些函数对于许多在Wi…

大模型综述总结--第一部分

1 目录 本文是学习https://github.com/le-wei/LLMSurvey/blob/main/assets/LLM_Survey_Chinese.pdf的总结&#xff0c;仅供学习&#xff0c;侵权联系就删 目录如下图 本次只总结一部分&#xff0c;刚学习有错请指出&#xff0c;VX关注晓理紫&#xff0c;关注后续。 2、概述…

红日靶场3

靶场链接&#xff1a;漏洞详情 在虚拟机的网络编辑器中添加两个仅主机网卡 信息搜集 端口扫描 外网机处于网端192.168.1.0/24中&#xff0c;扫描外网IP端口&#xff0c;开放了80 22 3306端口 80端口http服务&#xff0c;可以尝试登录网页 3306端口mysql服务&#xff0c;可…

linux卸载mysql8重装5

目录 背景操作卸载重装配置启动 背景 在linux&#xff08;阿里云ECS&#xff09;安装部署Hive时初始化Hive元数据库&#xff0c;遇到报错前一天两三小时没解决&#xff0c;问题定位为mysql&#xff0c;次日打算重装 操作 卸载 停止 MySQL 服务 systemctl stop mysql yum卸载…

ES6内置对象 - Map

Map&#xff08;Map对象保存键值对&#xff0c;键值均不限制类型&#xff09; 特点&#xff1a; 有序&#xff08;Set集合是无序的&#xff09;&#xff1b;键值对&#xff08;键可以是任意类型&#xff09;&#xff1b;键名不能重复&#xff08;如果重复&#xff0c;则覆盖&…

c编译器学习07:minilisp编译器改造(debug模式支持调试)

问题 原版的minilisp编译器不支持argv输入测试&#xff0c;不方便单步调试。 代码改造目标是既不改变原有程序的各种功能&#xff0c; 又能支持个人习惯的vs单步debug模式。 CMakeLists.txt变更 定义DEBUG宏 解决单步调试源码定位偏差问题 cmake_minimum_required(VERSION …

高级语言期末2012级B卷

1.编写函数&#xff0c;输出任意正整数n的位数&#xff08;n默认为存储十进制的整形变量&#xff09; 例如&#xff1a;正整数13&#xff0c;则输出2,&#xff1b;正整数3088&#xff0c;则输出4 #include <stdio.h>int func(int n) {int count0;while(n>0) {n/10;co…

免费的WP模板网站推荐

免费wordpress模板下载 高端大气上档次的免费wordpress主题&#xff0c;首页大图全屏显示经典风格的wordpress主题。 https://www.wpniu.com/themes/289.html wordpress免费企业主题 深蓝色经典实用的wordpress网站模板&#xff0c;用wordpress免费企业主题搭建网站。 http…

基于springboot+vue的安康旅游网站(前后端分离)

博主主页&#xff1a;猫头鹰源码 博主简介&#xff1a;Java领域优质创作者、CSDN博客专家、阿里云专家博主、公司架构师、全网粉丝5万、专注Java技术领域和毕业设计项目实战&#xff0c;欢迎高校老师\讲师\同行交流合作 ​主要内容&#xff1a;毕业设计(Javaweb项目|小程序|Pyt…

Day17_集合与数据结构(链表,栈和队列,Map,Collections工具类,二叉树,哈希表)

文章目录 Day17 集合与数据结构学习目标1 数据结构2 动态数组2.1 动态数组的特点2.2 自定义动态数组2.3 ArrayList与Vector的区别&#xff1f;2.4 ArrayList部分源码分析1、JDK1.6构造器2、JDK1.7构造器3、JDK1.8构造器4、添加与扩容5、删除元素6、get/set元素7、查询元素8、迭…