RabbitMQ5-死信队列

目录

死信的概念

死信的来源

死信实战

死信之TTl

死信之最大长度

死信之消息被拒

死信的概念

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

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

死信的来源

  • 消息 TTL 过期

    TTL是Time To Live的缩写, 也就是生存时间

  • 队列达到最大长度

    队列满了，无法再添加数据到 mq 中

  • 消息被拒绝

    (basic.reject 或 basic.nack) 并且 requeue=false

死信实战

死信之TTl

消费者 C1：

package com.qcby.sixin;import com.qcby.util.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;import java.util.HashMap;
import java.util.Map;/*** 死信队列 - 消费者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 Exception {Channel channel = RabbitMqUtils.getChannel();//声明死信和普通交换机 类型为 directchannel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);//声明死信队列String deadQueue = "dead-queue";channel.queueDeclare(deadQueue, false, false, false, null);//死信队列绑定：队列、交换机、路由键（routingKey）channel.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("等待接收消息........... ");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("Consumer01 接收到消息" + message);};channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {});}}

生产者：

package com.qcby.sixin;import com.qcby.util.RabbitMqUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;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);//设置消息的 TTL 时间 10sAMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();//该信息是用作演示队列个数限制for (int i = 1; i < 11; i++) {String message = "info" + i;channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());System.out.println("生产者发送消息:" + message);}}
}

消费者 C2：

package com.qcby.sixin;import com.qcby.util.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;public class Consumer02 {//死信交换机名称private static final String DEAD_EXCHANGE = "dead_exchange";public static void main(String[] args) throws Exception {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 -> {});}
}

启动 C1 ，之后关闭消费者，模拟其接收不到消息，再启动 Producer：

启动 C2 消费者，它消费死信队列里面的消息：

死信之最大长度

消费者 C1：

package com.qcby.sixin.two;import com.qcby.util.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;import java.util.HashMap;
import java.util.Map;/*** 死信队列 - 消费者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 Exception {Channel channel = RabbitMqUtils.getChannel();//声明死信和普通交换机 类型为 directchannel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);//声明死信队列String deadQueue = "dead-queue";channel.queueDeclare(deadQueue, false, false, false, null);//死信队列绑定：队列、交换机、路由键（routingKey）channel.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");//设置正常队列的长度限制params.put("x-max-length",6);//正常队列String normalQueue = "normal-queue";channel.queueDeclare(normalQueue, false, false, false, params);channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");System.out.println("等待接收消息........... ");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("Consumer01 接收到消息" + message);};channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {});}}

生产者：

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

消费者 C2：

package com.qcby.sixin.two;import com.qcby.util.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;public class Consumer02 {//死信交换机名称private static final String DEAD_EXCHANGE = "dead_exchange";public static void main(String[] args) throws Exception {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 -> {});}
}

死信之消息被拒

拒收消息 "info5"

消费者 C1：

package com.qcby.sixin.three;import com.qcby.util.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;import java.util.HashMap;
import java.util.Map;/*** 死信队列 - 消费者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 Exception {Channel channel = RabbitMqUtils.getChannel();//声明死信和普通交换机 类型为 directchannel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);//声明死信队列String deadQueue = "dead-queue";channel.queueDeclare(deadQueue, false, false, false, null);//死信队列绑定：队列、交换机、路由键（routingKey）channel.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("等待接收消息........... ");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);}};//开启手动应答channel.basicConsume(normalQueue, false, deliverCallback, consumerTag -> {});}}

生产者：

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

消费者 C2：

package com.qcby.sixin.three;import com.qcby.util.RabbitMqUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;public class Consumer02 {//死信交换机名称private static final String DEAD_EXCHANGE = "dead_exchange";public static void main(String[] args) throws Exception {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 -> {});}
}

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

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

相关文章

【项目初始化】

【项目初始化】

项目初始化 使用脚手架创建项目Vite创建项目推荐拓展 使用脚手架创建项目 Vite Vite 是一个现代的前端构建工具&#xff0c;它提供了极速的更新和开发体验&#xff0c;支持多种前端框架&#xff0c;如 Vue、React 等创建项目 pnpm create vuelatest推荐拓展
阅读更多...
一文读懂 Faiss：开启高维向量高效检索的大门

一文读懂 Faiss：开启高维向量高效检索的大门

一、引言 在大数据与人工智能蓬勃发展的当下&#xff0c;高维向量数据如潮水般涌现。无论是图像、音频、文本&#xff0c;还是生物信息领域&#xff0c;都离不开高维向量来精准刻画数据特征。然而&#xff0c;在海量的高维向量数据中进行快速、准确的相似性搜索&#xff0c;却…
阅读更多...
基于Django的Boss直聘IT岗位可视化分析系统的设计与实现

基于Django的Boss直聘IT岗位可视化分析系统的设计与实现

【Django】基于Django的Boss直聘IT岗位可视化分析系统的设计与实现&#xff08;完整系统源码开发笔记详细部署教程&#xff09;✅ 目录 一、项目简介二、项目界面展示三、项目视频展示 一、项目简介 该系统采用Python作为主要开发语言&#xff0c;利用Django这一高效、安全的W…
阅读更多...
python 语音识别

python 语音识别

目录 一、语音识别 二、代码实践 2.1 使用vosk三方库 2.2 使用SpeechRecognition 2.3 使用Whisper 一、语音识别 今天识别了别人做的这个app，觉得虽然是个日记app 但是用来学英语也挺好的，能进行语音识别，然后矫正语法，自己说的时候 ，实在不知道怎么说可以先乱说，然…
阅读更多...
栈和队列特别篇:栈和队列的经典算法问题

栈和队列特别篇:栈和队列的经典算法问题

图均为手绘,代码基于vs2022实现 系列文章目录 数据结构初探: 顺序表 数据结构初探:链表之单链表篇 数据结构初探:链表之双向链表篇 链表特别篇:链表经典算法问题 数据结构:栈篇 数据结构:队列篇 文章目录 系列文章目录前言一.有效的括号(leetcode 20)二.用队列实现栈(leetcode…
阅读更多...
使用 OpenResty 构建高效的动态图片水印代理服务20250127

使用 OpenResty 构建高效的动态图片水印代理服务20250127

使用 OpenResty 构建高效的动态图片水印代理服务 在当今数字化的时代&#xff0c;图片在各种业务场景中广泛应用。为了保护版权、统一品牌形象&#xff0c;动态图片水印功能显得尤为重要。然而&#xff0c;直接在后端服务中集成水印功能&#xff0c;往往会带来代码复杂度增加、…
阅读更多...
C++并行化编程

C++并行化编程

C并行化编程 C 简介 C 是一种静态类型的、编译式的、通用的、大小写敏感的、不规则的编程语言&#xff0c;支持过程化编程、面向对象编程和泛型编程。 C 被认为是一种中级语言&#xff0c;它综合了高级语言和低级语言的特点。 C 是由 Bjarne Stroustrup 于 1979 年在新泽西州美…
阅读更多...
Java开发vscode环境搭建

Java开发vscode环境搭建

1 几个名词 JDK Java Development Kit JRE Java Runtion Environment JVM JDK 包括 Compiler,debugger,JRE等。JRE包括JVM和Runtime Library。 2 配置环境 2.1 安装JDK 类比 C/C的 g工具 官网&#xff1a;https://www.oracle.com/java/technologies/downloads/ 根据自己使…
阅读更多...
pytorch基于FastText实现词嵌入

pytorch基于FastText实现词嵌入

FastText 是 Facebook AI Research 提出的 改进版 Word2Vec&#xff0c;可以&#xff1a; ✅ 利用 n-grams 处理未登录词 比 Word2Vec 更快、更准确 适用于中文等形态丰富的语言 完整的 PyTorch FastText 代码&#xff08;基于中文语料&#xff09;&#xff0c;包含&#xff1…
阅读更多...
riscv xv6学习笔记

riscv xv6学习笔记

文章目录 前言util实验sleeputil实验pingpongutil实验primesxv6初始化代码分析syscall实验tracesyscall实验sysinfoxv6内存学习笔记pgtbl实验Print a page tablepgtbl实验A kernel page table per processxv6 trap学习trap实验Backtracetrap实验Alarmlazy实验Lazy allocationxv…
阅读更多...
FFmpeg(7.1版本)编译：Ubuntu18.04交叉编译到ARM

FFmpeg(7.1版本)编译：Ubuntu18.04交叉编译到ARM

一、本地编译与交叉编译 1.本地编译 ① 本地编译&#xff1a;指的是在目标系统上进行编译的过程 , 生成的可执行文件和函数库只能在目标系统中使用。 如 : 在 Ubuntu中&#xff0c;本地编译的可执行文件只能在Ubuntu 系统中执行 , 无法在 Windows / Mac / Android / iOS 系…
阅读更多...
创新创业计划书|建筑垃圾资源化回收

创新创业计划书|建筑垃圾资源化回收

目录 第1部分 公司概况........................................................................ 1 第2部分 产品/服务...................................................................... 3 第3部分 研究与开发.................................................…
阅读更多...
如何利用天赋实现最大化的价值输出

如何利用天赋实现最大化的价值输出

这种文章&#xff0c;以我现在的实力很难写出来。所以需要引用一些视频。 上92高校容易吗 如果基于天赋努力&#xff0c;非常容易。 如果不是这样&#xff0c;非常非常难。 高考失败人生完蛋&#xff1f;复读考上交大&#xff0c;进入社会才发现学历只是一张纸&#xff0c;98…
阅读更多...
LigerUI在MVC模式下的响应原则

LigerUI在MVC模式下的响应原则

LigerUI是基于jQuery的UI框架&#xff0c;故他也是遵守jQuery的开发模式&#xff0c;但是也具有其特色的侦听函数&#xff0c;那么当LigerUI作为View层的时候&#xff0c;他所发送后端的必然是表单的数据&#xff0c;在此我们以俩个div为例&#xff1a; {Layout "~/View…
阅读更多...
【力扣】49.字母异位词分组

【力扣】49.字母异位词分组

AC截图 题目 思路 由于互为字母异位词的两个字符串包含的字母相同&#xff0c;因此对两个字符串分别进行排序之后得到的字符串一定是相同的&#xff0c;故可以将排序之后的字符串作为哈希表的键。 可以遍历strs&#xff0c;将其中每一个str排序&#xff0c;然后用unodered_ma…
阅读更多...
docker安装nacos2.2.4详解（含：nacos容器启动参数、环境变量、常见问题整理）

docker安装nacos2.2.4详解（含：nacos容器启动参数、环境变量、常见问题整理）

一、镜像下载 1、在线下载 在一台能连外网的linux上执行docker镜像拉取命令 docker pull nacos:2.2.4 2、离线包下载 两种方式&#xff1a; 方式一&#xff1a; -&#xff09;在一台能连外网的linux上安装docker执行第一步的命令下载镜像 -&#xff09;导出 # 导出镜像到…
阅读更多...
【图床配置】PicGO+Gitee方案

【图床配置】PicGO+Gitee方案

【图床配置】PicGOGitee方案 文章目录 【图床配置】PicGOGitee方案为啥要用图床图床是什么配置步骤下载安装PicGoPicGo配置创建Gitee仓库Typora中的设置 为啥要用图床 在Markdown中&#xff0c;图片默认是以路径的形式存在的&#xff0c;类似这样 可以看到这是本地路径&#x…
阅读更多...
【C++】类与对象（下）

【C++】类与对象（下）

&#x1f984; 个人主页: 小米里的大麦-CSDN博客 &#x1f38f; 所属专栏: 小米里的大麦——C专栏_CSDN博客 &#x1f381; 代码托管: 小米里的大麦的Gitee仓库 ⚙️ 操作环境: Visual Studio 2022 文章目录 1. 再谈构造函数1.1 构造函数体赋值1.2 初始化列表1.3 explicit 关键…
阅读更多...
SpringBoot笔记

SpringBoot笔记

1.创建 使用idea提供的脚手架创建springboot项目&#xff0c;选上需要的模块&#xff0c;会自动进行导包 打成jar包&#xff0c;之前直接用原生的maven打包的是一个瘦jar&#xff0c;不能直接跑&#xff0c;把服务器上部署的jar排除在外了&#xff0c;但是现在加上打包查件&am…
阅读更多...
Fiddler(一) - Fiddler简介_fiddler软件

Fiddler(一) - Fiddler简介_fiddler软件

文章目录 一、为什么选择Fiddler作为抓包工具? 二、什么是Fiddler?三、Fiddler使用界面简介四、延伸阅读 一、为什么选择Fiddler作为抓包工具? 抓包工具有很多&#xff0c;小到最常用的web调试工具firebug&#xff0c;大到通用性强大的抓包工具wireshark。为什么使用fid…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023