RabbitMQ (4)

RabbitMQ (4)

文章目录

  • 1. 死信的概念
  • 2. 死信的来源
  • 3. 死信代码案例
    • 3.1 TTL 过期时间
    • 3.2 超过队列最大长度
    • 3.3 拒绝消息

前言
 

上文我们已经学习完 交换机 ，知道了几个交换机的使用 ，下面我们来学习一下 死信队列

 

1. 死信的概念

 
先从概念解释上搞清楚这个定义，死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer 将消息投递到 broker 或者直接到queue 里了，consumer 从 queue 取出消息 进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。
 
应用场景：为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。还有比如说：用户在商城下单成功并点击去支付后在指定时间未支付时自动失效。

 

2. 死信的来源

 

  1. 消息 TTL 过期 : TTL 是 Time To Live 的缩写， TTL 就是 生存时间
  2. 队列达到最长长度 : 队列满了 ， 无法添加数据到 MQ 中
  3. 消息被拒绝 （basic.reject 或 basic.nack） 并且 requeue = false

 

3. 死信代码案例

 

这里 创建一个 direct 交换机 ，两个消费者 ， 一个生产者 ， 两个 队列 (一个为 消息队列 ， 一个为死信队列)

 
图:

在这里插入图片描述

 
代码 :

 

3.1 TTL 过期时间

 

生产者：

package org.example.seven;import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.example.utils.RabbitMQUtils;import java.io.IOException;
import java.util.concurrent.TimeoutException;// 生产者
public class Producer {// 普通交换机的名称public static final String NORMAL_EXCHANGE = "normal_exchange";public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();// 设置消息的过期时间 （TTL） 单位是 ms --> 设置消息的过期时间为 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());}}
}

 

消费者 c1 （启动之后关闭该消费者， 模拟其接受不到消息）

package org.example.seven;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import org.example.utils.RabbitMQUtils;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeoutException;public class Consumer01 {// 普通交换机的名称public static final String NORMAL_EXCHANGE = "normal_exchange";// 死信交换机的名称public static final String DEAD_EXCHANGE = "dead_change";// 普通队列的名称public static final String NORMAL_QUEUE = "normal_queue";// 死刑队列的名称public static final String DEAD_QUEUE = "dead_queue";public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();// 声明死信和普通交换机 , 类型为 direct （直接交换机）// 普通交换机channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);// 死信交换机channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);// 声明普通队列Map<String, Object> arguments = new HashMap<>();// 过期时间 10s 由生产者指定 更加灵活
//        arguments.put("x-message-ttl", 10000);// 正常的队列设置死信交换机arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);// 设置死信 路由键 (routingKey)arguments.put("x-dead-letter-routing-key", "lisi");// 声明队列channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);// 申明死刑队列channel.queueDeclare(DEAD_QUEUE, false, false, false, null);// 绑定普通的交换机与队列channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");// 绑定死信的交换机与死信的队列channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");System.out.println("等待接受消息");DeliverCallback deliverCallback = (tag, message) -> {System.out.println("C1 接收到的消息为: " + new String(message.getBody(), "UTF-8"));};channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, (tag) -> {});}
}

 

先启动消费者 C1，创建出队列，然后停止该 C1 的运行，则 C1 将无法收到队列的消息，无法收到的消息 10 秒后进入死信队列。启动生产者 producer 生产消息

在这里插入图片描述

 
c1 看完，我们在来写 c2 消费者 ，将进入到死信队列的消息 进行消费.

 
消费者c2

package org.example.seven;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import org.example.utils.RabbitMQUtils;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer02 {// 死信队列的名称public static final String DEAD_QUEUE = "dead_queue";public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();System.out.println("等待接受死信消息......");DeliverCallback deliverCallback = (tag, message) -> {System.out.println("C2 接收到的消息： " + new String(message.getBody(), "UTF-8"));};channel.basicConsume(DEAD_QUEUE, true, deliverCallback,(tag)->{});}
}

 
图:

在这里插入图片描述

 
看完 消息过期后 ，消息转发到 死信队列 被 c2 消费，下面我们来 尝试使用 死信最大长度 (队列满了，将多的消息转发到死信队列中）

 

3.2 超过队列最大长度

 

消息生产者代码 去掉 TTL 属性 ， 将 basicPublish 的第三个参数改为 null

 
生产者:

package org.example.seven;import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.example.utils.RabbitMQUtils;import java.io.IOException;
import java.util.concurrent.TimeoutException;// 生产者
public class Producer {// 普通交换机的名称public static final String NORMAL_EXCHANGE = "normal_exchange";public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();// 设置消息的过期时间 （TTL） 单位是 ms --> 设置消息的过期时间为 10s
//        AMQP.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());channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", null, message.getBytes());}}
}

 
c1 消费者 (启动之后关闭该消费者 模拟其接收不到消息)

package org.example.seven;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import org.example.utils.RabbitMQUtils;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeoutException;public class Consumer01 {// 普通交换机的名称public static final String NORMAL_EXCHANGE = "normal_exchange";// 死信交换机的名称public static final String DEAD_EXCHANGE = "dead_change";// 普通队列的名称public static final String NORMAL_QUEUE = "normal_queue";// 死刑队列的名称public static final String DEAD_QUEUE = "dead_queue";public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();// 声明死信和普通交换机 , 类型为 direct （直接交换机）// 普通交换机channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);// 死信交换机channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);// 声明普通队列Map<String, Object> arguments = new HashMap<>();// 过期时间 10s 由生产者指定 更加灵活
//        arguments.put("x-message-ttl", 10000);// 正常的队列设置死信交换机arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);// 设置死信 路由键 (routingKey)arguments.put("x-dead-letter-routing-key", "lisi");// 设置队列的限制 , 例如 发送 10 个消息 , 6 个为正常 , 4 个为死信arguments.put("x-max-length", 6);// 声明队列channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);// 申明死刑队列channel.queueDeclare(DEAD_QUEUE, false, false, false, null);// 绑定普通的交换机与队列channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");// 绑定死信的交换机与死信的队列channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");System.out.println("等待接受消息");DeliverCallback deliverCallback = (tag, message) -> {System.out.println("C1 接收到的消息为: " + new String(message.getBody(), "UTF-8"));};channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, (tag) -> {});}
}

 
注意:
 
这参数改变了(没有设置 ttl 时间，新增了 队列的 最大长度限制 为 6) ，所以 需要把原来队列删除

 
消费者c2 代码不变

package org.example.seven;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import org.example.utils.RabbitMQUtils;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer02 {// 死信队列的名称public static final String DEAD_QUEUE = "dead_queue";public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();System.out.println("等待接受死信消息......");DeliverCallback deliverCallback = (tag, message) -> {System.out.println("C2 接收到的消息： " + new String(message.getBody(), "UTF-8"));};channel.basicConsume(DEAD_QUEUE, true, deliverCallback,(tag)->{});}
}

 
效果:

在这里插入图片描述

 
这里 之所以要启动 c1 后在关闭，是为了展示 6个消息放到 普通队列 ，4个消息放到死信队列， 如果不这么做，发送的 10个消息 都会被 c1 消费 ，（消息发送到 队列后 , 立马 转发给 c1 导致 队列就不会达到 6 个 ，队列不会满 ，也就不会将消息转化给 死信队列).

在这里插入图片描述

 

3.3 拒绝消息

 
消息生产者 和 消费者 c2 与上面的代码一样

这里我们 拒绝 info7 消息 ，想要 拒绝 info7 消息，我们可以采用手动应答.

 
消费者c1

package org.example.seven;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import org.example.utils.RabbitMQUtils;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeoutException;public class Consumer01 {// 普通交换机的名称public static final String NORMAL_EXCHANGE = "normal_exchange";// 死信交换机的名称public static final String DEAD_EXCHANGE = "dead_change";// 普通队列的名称public static final String NORMAL_QUEUE = "normal_queue";// 死刑队列的名称public static final String DEAD_QUEUE = "dead_queue";public static void main(String[] args) throws IOException, TimeoutException {Channel channel = RabbitMQUtils.getChannel();// 声明死信和普通交换机 , 类型为 direct （直接交换机）// 普通交换机channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);// 死信交换机channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);// 声明普通队列Map<String, Object> arguments = new HashMap<>();// 过期时间 10s 由生产者指定 更加灵活
//        arguments.put("x-message-ttl", 10000);// 正常的队列设置死信交换机arguments.put("x-dead-letter-exchange", DEAD_EXCHANGE);// 设置死信 路由键 (routingKey)arguments.put("x-dead-letter-routing-key", "lisi");// 设置队列的限制 , 例如 发送 10 个消息 , 6 个为正常 , 4 个为死信
//        arguments.put("x-max-length", 6);// 声明队列channel.queueDeclare(NORMAL_QUEUE, false, false, false, arguments);// 申明死刑队列channel.queueDeclare(DEAD_QUEUE, false, false, false, null);// 绑定普通的交换机与队列channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");// 绑定死信的交换机与死信的队列channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");System.out.println("等待接受消息");DeliverCallback deliverCallback = (tag, message) -> {String msg = new String(message.getBody(), "UTF-8");if (msg.equals("info7")) {System.out.println("C1 接收到消息为: " + msg + " 此消息被 C1 拒绝");//requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中channel.basicReject(message.getEnvelope().getDeliveryTag(), false);} else {System.out.println("C1 接收到的消息为: " + msg);channel.basicAck(message.getEnvelope().getDeliveryTag(), false);}};
//        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, (tag) -> {});// 开启 手动应答channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, (tag) -> {});}
}

在这里插入图片描述

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

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

相关文章

【C++】mapset的底层结构 -- AVL树（高度平衡二叉搜索树）

【C++】mapset的底层结构 -- AVL树（高度平衡二叉搜索树）

前面我们对 map / multimap / set / multiset 进行了简单的介绍&#xff0c;可以发现&#xff0c;这几个容器有个共同点是&#xff1a;其底层都是按照二叉搜索树来实现的。 但是二叉搜索树有其自身的缺陷&#xff0c;假如往树中插入的元素有序或者接近有序&#xff0c;二叉搜索…
阅读更多...
漏洞复现-phpmyadmin_SQL注入 （CVE-2020-5504）

漏洞复现-phpmyadmin_SQL注入 （CVE-2020-5504）

phpmyadmin SQL注入 _&#xff08;CVE-2020-5504&#xff09; 漏洞信息 CVE-2020-5504sql注入漏洞Phpmyadmin 5.00以下 描述 ​ phpMyAdmin是Phpmyadmin团队的一套免费的、基于Web的MySQL数据库管理工具。该工具能够创建和删除数据库&#xff0c;创建、删除、修改数据库表&…
阅读更多...
Java学习 5.习题2.

Java学习 5.习题2.

练习题1&#xff1a;判断一个数字是偶数还是奇数 int num110;if(num1%20) {System.out.println("num1是一个偶数");}else{System.out.println("num1是一个奇数");} 练习题2&#xff1a;判断一个数是正数还是负数还是0 int num2-5;if(num2>0) {System.ou…
阅读更多...
yarn install 这个命令安装如何加速

yarn install 这个命令安装如何加速

yarn install 命令用来安装项目依赖&#xff0c;其速度受多种因素影响&#xff0c;如网络速度、npm/yarn包的源服务器、以及本地缓存等。以下是一些可能帮助你加速 yarn install 的方法&#xff1a; 1. 使用国内镜像 如果你在中国&#xff0c;可以使用淘宝的 npm 镜像&#x…
阅读更多...
【Javascript】json

【Javascript】json

目录 什么是json&#xff1f; 书写格式 json 序列化和反序列化 序列化 反序列化 什么是json&#xff1f; JSON(JavaScript Object Notation)是⼀种轻量级的数据交换格式&#xff0c;它基于JavaScript的⼀个⼦集&#xff0c;易于⼈的编写和阅读&#xff0c;也易于机器解析…
阅读更多...
栈、队列、矩阵的总结

栈、队列、矩阵的总结

栈的应用 括号匹配 表达式求值&#xff08;中缀&#xff0c;后缀&#xff09; 中缀转后缀&#xff08;机算&#xff09; 中缀机算 后缀机算 总结 特殊矩阵 对称矩阵的压缩存储 三角矩阵 三对角矩阵 稀疏矩阵的压缩存储
阅读更多...
OpenCV官方教程中文版 —— 傅里叶变换

OpenCV官方教程中文版 —— 傅里叶变换

OpenCV官方教程中文版 —— 傅里叶变换 前言一、原理二、Numpy 中的傅里叶变换三、OpenCV 中的傅里叶变换四、为什么拉普拉斯算子是高通滤波器&#xff1f; 前言 本小节我们将要学习&#xff1a; • 使用 OpenCV 对图像进行傅里叶变换 • 使用 Numpy 中 FFT&#xff08;快速…
阅读更多...
多线程线程池

多线程线程池

线程安全问题 当多个线程同时操作同一个共享资源的时候&#xff0c;可能会出现结果不符合预期的问题 解决安全问题 方式一: 同步代码块 作用: 把访问共享资源的核心代码给上锁&#xff0c;以此保证线程安全 格式: synchronized(同步锁) { 访问共享资源的核心代…
阅读更多...
unocss和tailwindcss css原子引擎

unocss和tailwindcss css原子引擎

第一种tailwindcss&#xff1a; tailwindcss官网 https://tailwindcss.com/docs/grid-column 基本介绍及优点分析 Tailwind CSS 中文文档 - 无需离开您的HTML&#xff0c;即可快速建立现代网站 PostCss 处理 Tailwind Css 基本流程 PostCSS - 是一个用 JavaScript 工具和插…
阅读更多...
67 内网安全-域横向smbwmi明文或hash传递

67 内网安全-域横向smbwmi明文或hash传递

#知识点1: windows2012以上版本默认关闭wdigest&#xff0c;攻击者无法从内存中获取明文密码windows2012以下版本如安装KB2871997补丁&#xff0c;同样也会导致无法获取明文密码针对以上情况&#xff0c;我们提供了4种方式解决此类问题 1.利用哈希hash传递(pth&#xff0c;ptk等…
阅读更多...
尚未解决：use_python()和use_virtualenv()的使用

尚未解决：use_python()和use_virtualenv()的使用

reticulate包为Python和R之间的互操作性提供了一套全面的工具。该包包含以下功能&#xff1a; 以多种方式从R调用Python&#xff0c;包括RMarkdown、获取Python脚本、导入Python模块以及在R会话中交互使用Python。 R和Python对象之间的转换&#xff08;例如&#xff0c;R和Pan…
阅读更多...
2023年中国调音台产业链、产量及市场规模分析[图]

2023年中国调音台产业链、产量及市场规模分析[图]

调音台是一种专业音频设备&#xff0c;用于混音、处理和控制音频信号。它通常用于音乐制作、现场演出、录音室以及广播等场景中。调音台允许用户调整不同声音来源的音频信号&#xff0c;使其在混音过程中达到理想的音质和平衡。调音台按信号出来方式可分为&#xff1a;模拟式调…
阅读更多...
UVa10976 Fractions Again?!（分数拆分）

UVa10976 Fractions Again?!（分数拆分）

1、题目 2、题意 输入正整数 k k k&#xff0c;找到所有正整数 x ≥ y x \ge y x≥y&#xff0c;使得 1 k 1 x 1 y \frac{1}{k} \frac{1}{x} \frac{1}{y} k1​x1​y1​。 3、分析 既然要求找出所有的 x , y x,y x,y&#xff0c;枚举对象自然是 x , y x,y x,y了。可…
阅读更多...
C# 基于腾讯云人脸核身和百度云证件识别技术相结合的 API 实现

C# 基于腾讯云人脸核身和百度云证件识别技术相结合的 API 实现

目录 腾讯云人脸核身技术 Craneoffice.net 采用的识别方式 1、活体人脸核身(权威库)&#xff1a; 2、活体人脸比对&#xff1a; 3、照片人脸核身(权威库)&#xff1a; 调用成本 百度云身份证识别 调用成本 相关结合点 核心代码 实现调用人脸核身API的示例 实现调用身…
阅读更多...
拓世大模型 | 立足行业所需，发力终端，缔造智能无限可能

拓世大模型 | 立足行业所需，发力终端，缔造智能无限可能

蒸汽机的发明为人类工业革命揭开序幕&#xff0c;引领了近现代产业变革。众所周知&#xff0c;而今AI技术的革命性突破&#xff0c;站在了时代舞台的中心&#xff0c;特别是大模型的崛起&#xff0c;无疑是第四次产业革命的焦点&#xff0c;它的地位可与当年的“蒸汽机”相提并…
阅读更多...
JAVA排序

JAVA排序

目录 再看各种排序前我们先了解一下什么叫 稳定性 插入排序&#xff1a; 希尔排序:(插入排序的优化) 测试插入排序和希尔排序(插入排序的优化)排序时间对比 选择排序: 选择排序的优化: 正确的 选择排序优化 快速排序(挖坑法:未优化) 快速排序的优化1 快速排序优化2 优化快速排序…
阅读更多...
server2012 通过防火墙开启局域网内限定IP进行远程桌面连接

server2012 通过防火墙开启局域网内限定IP进行远程桌面连接

我这里需要被远程桌面的电脑系统版本为windows server2012 1、打开允许远程连接设置 2、开启防火墙 3、设置允许“远程桌面应用”通过防火墙 勾选”远程桌面“ 3、入站规则设置 高级设置→入站规则→远程桌面-用户模式(TCP-In) 进入远程桌面属性的作用域——>远程IP地址—…
阅读更多...
实体店做商城小程序如何

实体店做商城小程序如何

互联网电商深入各个行业&#xff0c;传统线下店商家无论产品销售还是服务业&#xff0c;仅靠以往的经营模式&#xff0c;很难拓展到客户&#xff0c;老客流失严重&#xff0c;同时渠道单一&#xff0c;无法实现外地客户购物及线上客户赋能等。 入驻第三方平台有优势但也有不足…
阅读更多...
Spring Web MVC入门

Spring Web MVC入门

一&#xff1a;了解Spring Web MVC (1)关于Java开发 &#x1f31f;Java开发大多数场景是业务开发 比如说京东的业务就是电商卖货、今日头条的业务就推送新闻&#xff1b;快手的业务就是短视频推荐 (2)Spring Web MVC的简单理解 &#x1f497;Spring Web MVC&#xff1a;如何使…
阅读更多...
分布估计算法（Estimation of distribution algorithm，EDA）

分布估计算法（Estimation of distribution algorithm，EDA）

概论 分布估计算法&#xff08;Estimation of distribution algorithm&#xff0c;EDA&#xff09;是一种新兴的基于统计学原理的随机优化算法。 为什么要叫这个名字呢&#xff1f; 首先&#xff0c;“分布”指的就是概率分布。 其次&#xff0c;“估计”指的是这个概率分布…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023