RabbitMQ-消息队列：发布确认高级

18、发布确认高级

在生产环境中由于一些不明原因，导致 RabbitMQ 重启，在 RabbitMQ 重启期间生产者消息投递失败， 导致消息丢失，需要手动处理和恢复。于是，我们开始思考，如何才能进行 RabbitMQ 的消息可靠投递呢？

1、确认机制方案

在这里插入图片描述

2、代码架构图

在这里插入图片描述

在配置文件当中需要添加

spring.rabbitmq.publisher-confirm-type=correlated

  • NONE 值是禁用发布确认模式，是默认值。

  • CORRELATED 值是发布消息成功到交换器后会触发回调方法。

  • SIMPLE（最好不用） 值经测试有两种效果，其一效果和 CORRELATED 值一样会触发回调方法，其二在发布消息成功后使用 rabbitTemplate 调用 waitForConfirms 或 waitForConfirmsOrDie 方法等待 broker 节点返回发送结果，根据返回结果来判定下一步的逻辑，要注意的点是 waitForConfirmsOrDie 方法如果返回 false 则会关闭 channel，则接下来无法发送消息到 broker。(相当于单一发布)

3、正常发送接收代码

（1）配置类

在这里插入图片描述

在这里插入图片描述

（2）生产者

在这里插入图片描述

（3）消费者

在这里插入图片描述

（4）测试效果：

在这里插入图片描述

4、回调接口

在这里插入图片描述

如果是交换机出了问题，队列出不出问题，都无济于事，交换机出了问题，交换机的缓存就没什么作用了，所以需要通过回调解决交换机宕机的问题

（1）修改生产者代码，添加回调接口方法

在这里插入图片描述

（2）查看源码

在这里插入图片描述

实现内部接口函数的注意点：

在这里插入图片描述

交换机确认回调方法

在这里插入图片描述

（4）创建回调函数的三部曲

在这里插入图片描述

发送方设置回调函数的CorrelationData

在这里插入图片描述

（5）配置开启交换机回调服务

在这里插入图片描述

测试效果：

在这里插入图片描述

交换机宕掉

在这里插入图片描述

测试效果：

在这里插入图片描述

结果：出现找不到交换机的错误

队列宕掉

在这里插入图片描述

测试效果：

访问： http://localhost:8080/confirm/sendMessage/LBJ

在这里插入图片描述
在这里插入图片描述

结果：会丢失RK对应的队列消息

可以看到，发送了两条消息，第一条消息的 RoutingKey 为 “key1”，第二条消息的 RoutingKey 为 “key2”，两条消息都成功被交换机接收，也收到了交换机的确认回调，但消费者只收到了一条消息，因为第二条消息的 RoutingKey 与队列的 BindingKey 不一致，也没有其它队列能接收这个消息，所有第二条消息被直接丢弃了。丢弃的消息交换机是不知道的，需要解决告诉生产者消息传送失败。

代码实现

（1）配置类代码

@Configuration
public class ConfirmConfig {public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";public static final String CONFIRM_QUEUE_NAME = "confirm.queue";//声明业务 Exchange@Bean("confirmExchange")public DirectExchange confirmExchange() {return new DirectExchange(CONFIRM_EXCHANGE_NAME);}// 声明确认队列@Bean("confirmQueue")public Queue confirmQueue() {return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();}// 声明确认队列绑定关系@Beanpublic Binding queueBinding(@Qualifier("confirmQueue") Queue queue,@Qualifier("confirmExchange") DirectExchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("key1");}
}

（2）消息生产者的回调接口

@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback {/*** 交换机不管是否收到消息的一个回调方法** @param correlationData 消息相关数据* @param ack             交换机是否收到消息* @param cause           为收到消息的原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {String id = correlationData != null ? correlationData.getId() : "";if (ack) {log.info("交换机已经收到 id 为:{}的消息", id);} else {log.info("交换机还未收到 id 为:{}消息，原因:{}", id, cause);}}}

（3）消息生产者

@RestController
@RequestMapping("/confirm")
@Slf4j
public class ProducerController {public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";@Autowiredprivate RabbitTemplate rabbitTemplate;@Autowiredprivate MyCallBack myCallBack;//依赖注入 rabbitTemplate 之后再设置它的回调对象@PostConstructpublic void init() {rabbitTemplate.setConfirmCallback(myCallBack);}/*** 消息回调和退回** @param message*/@GetMapping("sendMessage/{message}")public void sendMessage(@PathVariable String message) {//指定消息 id 为 1CorrelationData correlationData1 = new CorrelationData("1");String routingKey = "key1";rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME, routingKey, message + routingKey, correlationData1);log.info(routingKey + "发送消息内容:{}", message + routingKey);CorrelationData correlationData2 = new CorrelationData("2");routingKey = "key2";rabbitTemplate.convertAndSend(CONFIRM_EXCHANGE_NAME, routingKey, message + routingKey, correlationData2);log.info(routingKey + "发送消息内容:{}", message + routingKey);}}

（4）消息消费者

@Component
@Slf4j
public class ConfirmConsumer {public static final String CONFIRM_QUEUE_NAME = "confirm.queue";@RabbitListener(queues = CONFIRM_QUEUE_NAME)public void receiveMsg(Message message) {String msg = new String(message.getBody());log.info("接受到队列 confirm.queue 消息:{}", msg);}}

5、回退消息

Mandatory 参数

在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如 果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。那么如何 让无法被路由的消息帮我想办法处理一下？最起码通知我一声，我好自己处理啊。通过设置 mandatory 参 数可以在当消息传递过程中不可达目的地时将消息返回给生产者。

(1)开启发布消息回退消息的核心配置

在这里插入图片描述

#消息退回
spring.rabbitmq.publisher-returns=true

（2）继承回调接口

在这里插入图片描述

（3）实现回调函数

在这里插入图片描述

 @Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {log.error("消息{}，被交换机{}退回，退回原因：{}，路由Key：{}",new String(message.getBody()),exchange,replyText,routingKey);}

测试效果1：

在这里插入图片描述

报错原因：没有注入自定义的回调函数到接口里面

在这里插入图片描述

测试效果2：

在这里插入图片描述

6、备份交换机

有了 mandatory 参数和回退消息，我们获得了对无法投递消息的感知能力，在生产者的消息无法被投递时发现并处理。但有时候，我们并不知道该如何处理这些无法路由的消息，最多打个日志，然后触发报警，再来手动处理。而通过日志来处理这些无法路由的消息是很不优雅的做法，特别是当生产者所在的服务有多台机器的时候，手动复制日志会更加麻烦而且容易出错。而且设置 mandatory 参数会增加生产者的复杂性，需要添加处理这些被退回的消息的逻辑。如果既不想丢失消息，又不想增加生产者的复杂性，该怎么做呢？

前面在设置死信队列的文章中，我们提到，可以为队列设置死信交换机来存储那些处理失败的消息，可是这些不可路由消息根本没有机会进入到队列，因此无法使用死信队列来保存消息。 在 RabbitMQ 中，有一种备份交换机的机制存在，可以很好的应对这个问题。

备份交换机可以理解为 RabbitMQ 中交换机的 “备胎”，当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常备份交换机的类型为 Fanout ，这样就能把所有消息都投递到与其绑定的队列中，然后我们①在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。当然，我们②还可以建立一个报警队列，用独立的消费者来进行监测和报警。

在这里插入图片描述

代码实现

（1）配置类中定义、声明交换机、队列

在这里插入图片描述

在这里插入图片描述

（2）按照架构图绑定交换机和队列

在这里插入图片描述

（3）设置备份交换机

在这里插入图片描述

@Configuration
public class ConfirmConfig {public static final String CONFIRM_EXCHANGE_NAME = "confirm.exchange";public static final String CONFIRM_QUEUE_NAME = "confirm.queue";public static final String BACKUP_EXCHANGE_NAME = "backup.exchange";public static final String BACKUP_QUEUE_NAME = "backup.queue";public static final String WARNING_QUEUE_NAME = "warning.queue";// 声明确认队列@Bean("confirmQueue")public Queue confirmQueue(){return QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();}//声明确认队列绑定关系@Beanpublic Binding queueBinding(@Qualifier("confirmQueue") Queue queue,@Qualifier("confirmExchange") DirectExchange exchange){return BindingBuilder.bind(queue).to(exchange).with("key1");}//声明备份 Exchange@Bean("backupExchange")public FanoutExchange backupExchange(){return new FanoutExchange(BACKUP_EXCHANGE_NAME);}//声明确认 Exchange 交换机的备份交换机@Bean("confirmExchange")public DirectExchange confirmExchange(){ExchangeBuilder exchangeBuilder =ExchangeBuilder.directExchange(CONFIRM_EXCHANGE_NAME).durable(true)
//设置该交换机的备份交换机
.withArgument("alternate-exchange", BACKUP_EXCHANGE_NAME);return (DirectExchange)exchangeBuilder.build();}// 声明警告队列@Bean("warningQueue")public Queue warningQueue(){return QueueBuilder.durable(WARNING_QUEUE_NAME).build();}// 声明报警队列绑定关系@Beanpublic Binding warningBinding(@Qualifier("warningQueue") Queue queue,@Qualifier("backupExchange") FanoutExchange 
backupExchange){return BindingBuilder.bind(queue).to(backupExchange);}// 声明备份队列@Bean("backQueue")public Queue backQueue(){return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();}// 声明备份队列绑定关系@Beanpublic Binding backupBinding(@Qualifier("backQueue") Queue queue,@Qualifier("backupExchange") FanoutExchange backupExchange){return BindingBuilder.bind(queue).to(backupExchange);}
}

（4）报警消费者（转发消费者自己拓展）

在这里插入图片描述

测试效果：

测试之前：

去RabbitMQ的web管理界面删除已经建立的交换机，否则会和将要建立的交换机和备用交换机起冲突

重新启动项目的时候需要把原来的 confirm.exchange 删除因为我们修改了其绑定属性

在这里插入图片描述

备用交换机成功路由

在这里插入图片描述

小结

mandatory 参数与备份交换机可以一起使用的时候，如果两者同时开启，消息究竟何去何从？谁优先 级高，经过上面结果显示答案是备份交换机优先级高。

在这里插入图片描述

回退消息的优先级低

RabbitMQ-消息队列：发布确认高级 到此完结，笔者归纳、创作不易，大佬们给个3连再起飞吧

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

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

相关文章

ARM服务器部署Kafka集群

ARM服务器部署Kafka集群

安装前必备的条件是: (1)安装jdk(提供环境); (2)安装zookeeper(注册kafka信息); 需要这方面信息的可以查看我之前写的文档; 一.下载安装包 Kafka官网下载地址 Apache Kafka 根据自己需要下载相应的版本 目前最新的版本是3.6.1。 二.解压安装包 服务器上传下载好的kafk…
阅读更多...
《Linux C编程实战》笔记：信号量

《Linux C编程实战》笔记：信号量

信号量在操作系统的书里一般都有介绍，这里就只写书上说的了。 信号量是一个计数器，常用于处理进程或线程的同步问题，特别是对临界资源访问的同步。临界资源可以简单地理解为在某一时刻只能由一个进程或线程进行操作的资源，这里的…
阅读更多...
【零基础入门】什么样的人适合学网络安全？该怎么学？

【零基础入门】什么样的人适合学网络安全？该怎么学？

有很多想要转行网络安全或者选择网络安全专业的人在进行决定之前一定会有的问题： 什么样的人适合学习网络安全？我适不适合学习网络安全？ 到底什么样的人适合学习网络安全呢？（需要什么前提条件呢） 开门见山…
阅读更多...
【Appium UI自动化】pytest运行常见错误解决办法

【Appium UI自动化】pytest运行常见错误解决办法

通过Appium工具录制代码在pycharm上运行报错： 错误一： 1.提示 setup() 方法运行 error failed 解决办法：未创建 init __ 方法，创建一个空的__init.py文件就解决了。 原因： 错误二： 2.运行代码&#xff…
阅读更多...
Linux之ACL权限管理

Linux之ACL权限管理

文章目录 1.ACL权限介绍二、操作步骤1. 添加测试目录、用户、组，并将用户添加到组2. 修改目录的所有者和所属组3. 设定权限4. 为临时用户分配权限5. 验证acl权限6. 控制组的acl权限 1.ACL权限介绍 每个项目成员有一个自己的项目目录，对自己的目录有完全…
阅读更多...
构建生物医学知识图谱from zero to hero （4）：通过Neo4j构建知识图谱

构建生物医学知识图谱from zero to hero （4）：通过Neo4j构建知识图谱

图数据库是一种专门用于存储图形数据的 NoSQL 数据库。与传统的关系型数据库和其他 NoSQL 数据库不同，图数据库利用图形数据模型来存储和管理数据。图形数据模型由节点和边组成，节点代表实体，边代表实体之间的关系。例如，在社交网络中，用户可以表示为节点，朋友关系可以表…
阅读更多...
xff注入 [CISCN2019 华东南赛区]Web111

xff注入 [CISCN2019 华东南赛区]Web111

打开题目 看见smarty 想到模板注入 又看见ip 想到xff注入 一般情况下输入{$smarty.version}就可以看到返回的smarty的版本号。该题目的Smarty版本是3.1.30 在Smarty3的官方手册里有以下描述: Smarty已经废弃{php}标签，强烈建议不要使用。在Smarty 3.1&#xff…
阅读更多...
C# OpenVINO 百度PaddleSeg实时人像抠图PP-MattingV2

C# OpenVINO 百度PaddleSeg实时人像抠图PP-MattingV2

目录 效果 项目 代码 下载 C# OpenVINO 百度PaddleSeg实时人像抠图PP-MattingV2 效果 项目 代码 using OpenCvSharp; using Sdcb.OpenVINO; using System; using System.Diagnostics; using System.Drawing; using System.Security.Cryptography; using System.Text; us…
阅读更多...
SparkSQL学习03-数据读取与存储

SparkSQL学习03-数据读取与存储

文章目录 1 数据的加载1.1 方式一：spark.read.format1.1.1读取json数据1.1.2 读取jdbc数据 1.2 方式二：spark.read.xxx1.2.1 读取json数据1.2.2 读取csv数据1.2.3 读取txt数据1.2.4 读取parquet数据1.2.5 读取orc数据1.2.6 读取jdbc数据 2 数据的保存2.1…
阅读更多...
SmartX 携手 openGauss 社区发布联合方案评测与性能最佳实践

SmartX 携手 openGauss 社区发布联合方案评测与性能最佳实践

近日，北京志凌海纳科技有限公司（以下简称 “SmartX”）携手 openGauss 社区完成了 openGauss 数据库基于 SmartX 超融合平台（SMTX OS）和 SmartX 分布式存储平台（SMTX ZBS）的性能测试和调优。 结果…
阅读更多...
MySQL数据库基础（十三）：关系型数据库三范式介绍

MySQL数据库基础（十三）：关系型数据库三范式介绍

文章目录 关系型数据库三范式介绍 一、什么是三范式 二、数据冗余 三、范式的划分 四、一范式 五、二范式 六、三范式 七、总结 关系型数据库三范式介绍 一、什么是三范式 设计关系数据库时，遵从不同的规范要求，设计出合理的关系型数据库&…
阅读更多...
紫光同创初使用

紫光同创初使用

芯片PGC2KG-6LPG144 1、安装好软件接，加载license,有两个，与电脑MAC地址绑定的 2、正常使用后，新建个工程，配置管脚Tools→UCE 3、程序中有些信号被软件认为是时钟信号，会报错（时钟输入I0约束在非专用时钟…
阅读更多...
【LeetCode刷题笔记】242.有效的字母异位词

【LeetCode刷题笔记】242.有效的字母异位词

创作不易&#xff0c;本篇文章如果帮助到了你&#xff0c;还请点赞 关注支持一下♡>&#x16966;<)!! 主页专栏有更多知识&#xff0c;如有疑问欢迎大家指正讨论&#xff0c;共同进步&#xff01; 更多算法知识专栏&#xff1a;算法分析&#x1f525; 给大家跳段街舞感谢…
阅读更多...
Spring基础之AOP和代理模式

Spring基础之AOP和代理模式

文章目录 理解AOPAOP的实现原理 AOP代理模式静态代理动态代理1-JDK动态代理2-CGLIB动态代理 总结 理解AOP OOP - - Object Oriented Programming 面向对象编程 AOP - - Aspect Oriented Programming 面向切面编程 AOP是Spring提供的关键特性之一。AOP即面向切面编程&#xff0…
阅读更多...
Jenkins邮件通知配置（7）

Jenkins邮件通知配置（7）

1、安装插件&#xff1a; Email Extension&#xff0c;Email Extension Template&#xff0c;这两个插件可以帮助我们进行邮件的编写发送以及格式化 2、配置jenkins中链接腾讯企业邮箱 先配置发送服务&#xff0c;然后在具体工程中设置接收者 基础信息&#xff1a; POP3/S…
阅读更多...
SWIFT：自我认知微调

SWIFT：自我认知微调

文档：https://github.com/modelscope/swift/blob/main/docs/source/LLM/%E8%87%AA%E6%88%91%E8%AE%A4%E7%9F%A5%E5%BE%AE%E8%B0%83%E6%9C%80%E4%BD%B3%E5%AE%9E%E8%B7%B5.md ​​​​​​代码： Swift是如何把自我认知数据集融合到训练集中呢？ 1：相关的3个参数
阅读更多...
企业级大数据安全架构（十一）Kerberos接入dophinscheduler

企业级大数据安全架构（十一）Kerberos接入dophinscheduler

作者&#xff1a;楼高 建议将dophinscheduler集成到Ambari安装部署&#xff0c;在Ambari上面开启kerberos 1.安装准备 编译 从GitHub获取dolphinscheduler-1.3.9源码 git clone https://github.com/apache/dolphinscheduler.git -b 1.3.9-releasehttps://github.com/apache/…
阅读更多...
多输入回归预测|GWO-CNN-LSTM|灰狼算法优化的卷积-长短期神经网络回归预测(Matlab)

多输入回归预测|GWO-CNN-LSTM|灰狼算法优化的卷积-长短期神经网络回归预测(Matlab)

目录 一、程序及算法内容介绍&#xff1a; 基本内容&#xff1a; 亮点与优势&#xff1a; 二、实际运行效果&#xff1a; 三、算法介绍&#xff1a; 灰狼优化算法&#xff1a; 卷积神经网络-长短期记忆网络&#xff1a; 四、完整程序下载&#xff1a; 一、程序及算法内容…
阅读更多...
ChatGPT/GPT4科研应用与AI绘图及论文写作

ChatGPT/GPT4科研应用与AI绘图及论文写作

2023年随着OpenAI开发者大会的召开&#xff0c;最重磅更新当属GPTs&#xff0c;多模态API&#xff0c;未来自定义专属的GPT。微软创始人比尔盖茨称ChatGPT的出现有着重大历史意义&#xff0c;不亚于互联网和个人电脑的问世。360创始人周鸿祎认为未来各行各业如果不能搭上这班车…
阅读更多...
重看LeakCanary

重看LeakCanary

LeakCanary是我很久之前看的东西了&#xff0c;我当时侯对它的印象就是它可以用来检测内存泄漏&#xff0c;具体原理就是将弱引用对象延迟个5s然后看是否被回收,如果没有被回收,那么就说明发生了内存泄漏,其他的也没有仔细地看 现在就详细地梳理一遍这个流程&#xff1a; 1.L…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023