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,一经查实,立即删除!

相关文章

Redis 与 MySQL 一致性 实现方案

正常情况下的流程是&#xff1a;请求来了&#xff0c;先检查 Redis 有没有数据&#xff0c;有返回&#xff1b;没有便查询 MySQL 然后 放入 Redis。 此时&#xff0c;如果 MySQL 的数据发生了变化&#xff0c;所以需要同步到 Redis 中。 解决方法&#xff1a;MySQL 中的数据更新…

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

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

漏洞复现-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.

练习题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…

vantUI(Tabbar标签页)浏览器返回上一页的失效问题

在开发中遇到这样一个问题&#xff0c;由页面1切换到页面2&#xff0c;再点击浏览器的回退&#xff0c;无法回退到页面1。 开始以为是路由配置的有问题&#xff0c;但是子页面可以正常回退&#xff0c;因为replace只是替换路由&#xff0c;而不会往history栈中记录路由&#x…

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

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

Android 13.0 系统多个播放器app时,设置默认播放器

1.概述 在13.0的系统产品开发中,对于在系统中有多个播放器的时候,这时候如果调用代码打开播放器,会出现多个播放器列表让用户 选择启动哪个播放器,所以产品开发需求需要设置默认播放器,当打开播放器的时候,就直接打开播放器就可以了,所以就需要 了解查询播放器列表流程,…

【Javascript】json

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

栈、队列、矩阵的总结

栈的应用 括号匹配 表达式求值&#xff08;中缀&#xff0c;后缀&#xff09; 中缀转后缀&#xff08;机算&#xff09; 中缀机算 后缀机算 总结 特殊矩阵 对称矩阵的压缩存储 三角矩阵 三对角矩阵 稀疏矩阵的压缩存储

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

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

多线程线程池

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

202310-MetaPhlAn4安装和使用方法-Anaconda3- centos9 stream

MetaPhlAn 4是一种基于DNA序列的微生物组分析工具&#xff0c;它能够从宏基因组测序数据中识别和分离微生物的组成。以下是安装和使用MetaPhlAn 4的步骤&#xff1a; 安装MetaPhlAn 4&#xff1a; 裸机环境&#xff0c;手动安装&#xff08; 1. 安装依赖项&#xff1a; Met…

unocss和tailwindcss css原子引擎

第一种tailwindcss&#xff1a; tailwindcss官网 https://tailwindcss.com/docs/grid-column 基本介绍及优点分析 Tailwind CSS 中文文档 - 无需离开您的HTML&#xff0c;即可快速建立现代网站 PostCss 处理 Tailwind Css 基本流程 PostCSS - 是一个用 JavaScript 工具和插…

SM3加密udf

SM3加密udf maven xml <dependencies> <!-- 配置日志 --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.5</version> </dep…

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

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

读书笔记:c++对话系列,Visitor模式

需要在类体系中加入一个新的虚函数&#xff0c;但不允许改动。 Visitor模式&#xff0c;核心是&#xff1a;将操作作为数据对象传递给类体系预留的Accept函数。 class Personnel{ public:virtual void Pay () 0;virtual void Promote() 0;virtual void Accept(Visitor &…

缓冲channel和限制goroutine并发数

文章目录 3.7.1 用作计数信号量3.7.2 使用缓存channelsync.WaitGroup限制并发数&#xff08;类似上小节&#xff09; 要限制住goroutine的并发, 一定要阻塞住main的goroutine! 一定要阻塞住main的goroutine! 一定要阻塞住main的goroutine! 可以看最后一个例子。 由于带缓冲cha…

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

reticulate包为Python和R之间的互操作性提供了一套全面的工具。该包包含以下功能&#xff1a; 以多种方式从R调用Python&#xff0c;包括RMarkdown、获取Python脚本、导入Python模块以及在R会话中交互使用Python。 R和Python对象之间的转换&#xff08;例如&#xff0c;R和Pan…

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

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

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了。可…