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集群

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

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

信号量在操作系统的书里一般都有介绍,这里就只写书上说的了。 信号量是一个计数器,常用于处理进程或线程的同步问题,特别是对临界资源访问的同步。临界资源可以简单地理解为在某一时刻只能由一个进程或线程进行操作的资源,这里的…

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

有很多想要转行网络安全或者选择网络安全专业的人在进行决定之前一定会有的问题: 什么样的人适合学习网络安全?我适不适合学习网络安全? 到底什么样的人适合学习网络安全呢?(需要什么前提条件呢) 开门见山…

订单到期关闭如何实现(延迟场景处理)

文章目录 概述种处理方案任务调度实现(定时任务)基于redis 如何实现1redis过期key实现(键通知机制)基于redis延迟队列Redisson实现一个延迟队列 基于MQ的延迟队列实现 概述 订单30分钟未支付自动取消怎么实现 日常开发中,我们经常遇到这种业…

2. Linux进程管理

2. 进程管理 2.1 Linux进程 进程是程序执行时的一个动态实体,包含程序计数器、全部CPU 寄存器的值和进程堆栈中存储着的一些临时数据,如子程序参数、返回地址及变量等,反映的是当前处理器的活动状态。 而程序是仅包含指令和数据的一段静态代…

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

通过Appium工具录制代码在pycharm上运行报错: 错误一: 1.提示 setup() 方法运行 error failed 解决办法:未创建 init __ 方法,创建一个空的__init.py文件就解决了。 原因: 错误二: 2.运行代码&#xff…

Linux之ACL权限管理

文章目录 1.ACL权限介绍二、操作步骤1. 添加测试目录、用户、组,并将用户添加到组2. 修改目录的所有者和所属组3. 设定权限4. 为临时用户分配权限5. 验证acl权限6. 控制组的acl权限 1.ACL权限介绍 每个项目成员有一个自己的项目目录,对自己的目录有完全…

构建生物医学知识图谱from zero to hero (4):通过Neo4j构建知识图谱

图数据库是一种专门用于存储图形数据的 NoSQL 数据库。与传统的关系型数据库和其他 NoSQL 数据库不同,图数据库利用图形数据模型来存储和管理数据。图形数据模型由节点和边组成,节点代表实体,边代表实体之间的关系。例如,在社交网络中,用户可以表示为节点,朋友关系可以表…

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 效果 项目 代码 using OpenCvSharp; using Sdcb.OpenVINO; using System; using System.Diagnostics; using System.Drawing; using System.Security.Cryptography; using System.Text; us…

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 社区完成了 openGauss 数据库基于 SmartX 超融合平台(SMTX OS)和 SmartX 分布式存储平台(SMTX ZBS)的性能测试和调优。 结果…

JavaScript中的可选链——通过示例解释

JavaScript开发经常涉及导航嵌套对象,这可能很麻烦且容易出错,特别是在处理null或undefined值时。可选链是现代JavaScript语法中的一个改革性特性。 在本文中,我们将通过实际示例探讨可选链,演示它如何简化代码并使开发更加高效。…

MySQL数据库基础(十三):关系型数据库三范式介绍

文章目录 关系型数据库三范式介绍 一、什么是三范式 二、数据冗余 三、范式的划分 四、一范式 五、二范式 六、三范式 七、总结 关系型数据库三范式介绍 一、什么是三范式 设计关系数据库时,遵从不同的规范要求,设计出合理的关系型数据库&…

代码随想录算法训练营第五十九天|

583. 两个字符串的删除操作 本题和动态规划:115.不同的子序列 相比,其实就是两个字符串都可以删除了,情况虽说复杂一些,但整体思路是不变的。 代码随想录 class Solution {public int minDistance(String word1, String word2) {…

流畅的Python(十一)-从协议到抽象基类

一、核心要义 主要讨论Python中的接口,所谓接口就是类实现或继承的一套公开(按照定义,受保护的属性和私有属性不在接口中)属性和方法,包括特殊方法,如__getitem__或__add__等。Python有两套规范接口的方式: 1. 鸭子类型和协议,这…

几种后端开发中常用的语言。

几种后端开发中常用的语言。 C/C 语言 C 语言最初是用于系统开发工作,特别是组成操作系统的程序。由于 C 语言所产生的代码运行速度与汇编语言编写的代码运行速度几乎一样,所以采用 C 语言作为系统开发语言。目前,C 语言是最广泛使用的系统…

MongoDB聚合运算符:$atan2

$atan2用来计算反正切&#xff0c;返回指定表达式的反正切值&#xff0c;与$antan的区别主要是参数不同。 语法 { $atan2: [<expression1>, <expression1>] }<expression>为可被解析为数值的表达式$atan2返回弧度&#xff0c;使用$radiansToDegrees运算符可…

数据结构与算法-常用排序算法

一、常用排序说明 当涉及排序算法时&#xff0c;理解每个算法的工作原理、时间复杂度和空间复杂度是至关重要的。下面对常用排序算法进行详细说明&#xff1a; 1、冒泡排序&#xff08;Bubble Sort&#xff09;&#xff1a; 工作原理&#xff1a;比较相邻的元素并交换&am…

python bug与debug

一、什么是bug&#xff08;软件缺陷&#xff09;&#xff1f; 产品说明书中规定要做的事情&#xff0c;而软件没有实现。 产品说明书中规定不要做的事情&#xff0c;而软件确实现了。 产品说明书中没有提到过的事情&#xff0c;而软件确实现了。 产品说明书中没有提到但是必…