RabbitMQ保证消息的可靠性

1. 问题引入

消息从发送,到消费者接收,会经理多个过程:
在这里插入图片描述

其中的每一步都可能导致消息丢失,常见的丢失原因包括:

  • 发送时丢失:
    • 生产者发送的消息未送达exchange
    • 消息到达exchange后未到达queue
  • MQ宕机,queue将消息丢失
  • consumer接收到消息后未消费就宕机

针对这些问题,RabbitMQ分别给出了解决方案:

  • 生产者确认机制
  • mq持久化
  • 消费者确认机制
  • 失败重试机制

2. 生产者消息确认

RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。这种机制必须给每个消息指定一个唯一ID。消息发送到MQ以后,会返回一个结果给发送者,表示消息是否处理成功。

返回结果有两种方式:

  • publisher-confirm,发送者确认
    • 消息成功投递到交换机,返回ack
    • 消息未投递到交换机,返回nack
  • publisher-return,发送者回执
    • 消息投递到交换机了,但是没有路由到队列。返回ACK,及路由失败原因。
      在这里插入图片描述

注意:
在这里插入图片描述

2.1 修改配置

首先,修改publisher服务中的application.yml文件,添加下面的内容:

spring:rabbitmq:publisher-confirm-type: correlatedpublisher-returns: truetemplate:mandatory: true

说明:

  • publish-confirm-type:开启publisher-confirm,这里支持两种类型:
    • simple:同步等待confirm结果,直到超时
    • correlated:异步回调,定义ConfirmCallback,MQ返回结果时会回调这个ConfirmCallback
  • publish-returns:开启publish-return功能,同样是基于callback机制,不过是定义ReturnCallback
  • template.mandatory:定义消息路由失败时的策略。true,则调用ReturnCallback;false:则直接丢弃消息

2.2 定义Return回调

每个RabbitTemplate只能配置一个ReturnCallback,因此需要在项目加载时配置:

修改publisher服务,添加一个:

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;@Slf4j
@Configuration
public class CommonConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取RabbitTemplateRabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);// 设置ReturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {// 投递失败,记录日志log.info("消息发送失败,应答码{},原因{},交换机{},路由键{},消息{}",replyCode, replyText, exchange, routingKey, message.toString());// 如果有业务需要,可以重发消息});}
}

2.3 定义ConfirmCallback

ConfirmCallback可以在发送消息时指定,因为每个业务处理confirm成功或失败的逻辑不一定相同。

在publisher服务的测试类中,定义一个单元测试方法:

public void testSendMessage2SimpleQueue() throws InterruptedException {// 1.消息体String message = "hello, spring amqp!";// 2.全局唯一的消息ID,需要封装到CorrelationData中CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());// 3.添加callbackcorrelationData.getFuture().addCallback(result -> {if(result.isAck()){// 3.1.ack,消息成功log.debug("消息发送成功, ID:{}", correlationData.getId());}else{// 3.2.nack,消息失败log.error("消息发送失败, ID:{}, 原因{}",correlationData.getId(), result.getReason());}},ex -> log.error("消息发送异常, ID:{}, 原因{}",correlationData.getId(),ex.getMessage()));// 4.发送消息rabbitTemplate.convertAndSend("task.direct", "task", message, correlationData);// 休眠一会儿,等待ack回执Thread.sleep(2000);
}

3. 消息持久化

生产者确认可以确保消息投递到RabbitMQ的队列中,但是消息发送到RabbitMQ以后,如果突然宕机,也可能导致消息丢失。

要想确保消息在RabbitMQ中安全保存,必须开启消息持久化机制。

  • 交换机持久化
  • 队列持久化
  • 消息持久化

3.1 交换机持久化

RabbitMQ中交换机默认是非持久化的,mq重启后就丢失。

SpringAMQP中可以通过代码指定交换机持久化:

@Bean
public DirectExchange simpleExchange(){// 三个参数:交换机名称、是否持久化、当没有queue与其绑定时是否自动删除return new DirectExchange("simple.direct", true, false);
}

事实上,默认情况下,由SpringAMQP声明的交换机都是持久化的。

可以在RabbitMQ控制台看到持久化的交换机都会带上D的标示:
在这里插入图片描述

3.2 队列持久化

RabbitMQ中队列默认是非持久化的,mq重启后就丢失。

SpringAMQP中可以通过代码指定交换机持久化:

@Bean
public Queue simpleQueue(){// 使用QueueBuilder构建队列,durable就是持久化的return QueueBuilder.durable("simple.queue").build();
}

事实上,默认情况下,由SpringAMQP声明的队列都是持久化的。

可以在RabbitMQ控制台看到持久化的队列都会带上D的标示:
在这里插入图片描述

3.3 消息持久化

利用SpringAMQP发送消息时,可以设置消息的属性(MessageProperties),指定delivery-mode:

  • 1:非持久化
  • 2:持久化

用Java代码指定:
在这里插入图片描述

默认情况下,SpringAMQP发出的任何消息都是持久化的,不用特意指定。

4. 消费者消息确认

RabbitMQ是阅后即焚机制,RabbitMQ确认消息被消费者消费后会立刻删除。

而RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的:消费者获取消息后,应该向RabbitMQ发送ACK回执,表明自己已经处理消息。

设想这样的场景:

  • 1)RabbitMQ投递消息给消费者
  • 2)消费者获取消息后,返回ACK给RabbitMQ
  • 3)RabbitMQ删除消息
  • 4)消费者宕机,消息尚未处理

这样,消息就丢失了。因此消费者返回ACK的时机非常重要。

而SpringAMQP则允许配置三种确认模式:

•manual:手动ack,需要在业务代码结束后,调用api发送ack。

•auto:自动ack,由spring监测listener代码是否出现异常,没有异常则返回ack;抛出异常则返回nack

•none:关闭ack,MQ假定消费者获取消息后会成功处理,因此消息投递后立即被删除

由此可知:

  • none模式下,消息投递是不可靠的,可能丢失
  • auto模式类似事务机制,出现异常时返回nack,消息回滚到mq;没有异常,返回ack
  • manual:自己根据业务情况,判断什么时候该ack

一般,我们都是使用默认的auto即可。

4.1 none模式

修改consumer服务的application.yml文件,添加下面内容:

spring:rabbitmq:listener:simple:acknowledge-mode: none # 关闭ack

修改consumer服务的SpringRabbitListener类中的方法,模拟一个消息处理异常:

@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String msg) {log.info("消费者接收到simple.queue的消息:【{}】", msg);// 模拟异常System.out.println(1 / 0);log.debug("消息处理完成!");
}

测试可以发现,当消息处理抛异常时,消息依然被RabbitMQ删除了。

4.2 auto模式

再次把确认机制修改为auto:

spring:rabbitmq:listener:simple:acknowledge-mode: auto # 关闭ack

在异常位置打断点,再次发送消息,程序卡在断点时,可以发现此时消息状态为unack(未确定状态):
在这里插入图片描述

抛出异常后,因为Spring会自动返回nack,所以消息恢复至Ready状态,并且没有被RabbitMQ删除:
在这里插入图片描述

5. 消费失败重试机制

当消费者出现异常后,消息会不断requeue(重入队)到队列,再重新发送给消费者,然后再次异常,再次requeue,无限循环,导致mq的消息处理飙升,带来不必要的压力:
在这里插入图片描述

怎么办呢?

5.1 本地重试

我们可以利用Spring的retry机制,在消费者出现异常时利用本地重试,而不是无限制的requeue到mq队列。

修改consumer服务的application.yml文件,添加内容:

spring:rabbitmq:listener:simple:retry:enabled: true # 开启消费者失败重试initial-interval: 1000 # 初识的失败等待时长为1秒multiplier: 1 # 失败的等待时长倍数,下次等待时长 = multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态;false有状态。如果业务中包含事务,这里改为false

重启consumer服务,重复之前的测试。可以发现:

  • 在重试3次后,SpringAMQP会抛出异常AmqpRejectAndDontRequeueException,说明本地重试触发了
  • 查看RabbitMQ控制台,发现消息被删除了,说明最后SpringAMQP返回的是ack,mq删除消息了

结论:

  • 开启本地重试时,消息处理过程中抛出异常,不会requeue到队列,而是在消费者本地重试
  • 重试达到最大次数后,Spring会返回ack,消息会被丢弃

5.2 失败策略

在之前的测试中,达到最大重试次数后,消息会被丢弃,这是由Spring内部机制决定的。

在开启重试模式后,重试次数耗尽,如果消息依然失败,则需要有MessageRecovery接口来处理,它包含三种不同的实现:

  • RejectAndDontRequeueRecoverer:重试耗尽后,直接reject,丢弃消息。默认就是这种方式

  • ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队

  • RepublishMessageRecoverer:重试耗尽后,将失败消息投递到指定的交换机

比较优雅的一种处理方案是RepublishMessageRecoverer,失败后将消息投递到一个指定的,专门存放异常消息的队列,后续由人工集中处理。

1)在consumer服务中定义处理失败消息的交换机和队列

@Bean
public DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue(){return new Queue("error.queue", true);
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
}

2)定义一个RepublishMessageRecoverer,关联队列和交换机

@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}

完整代码:

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;@Configuration
public class ErrorMessageConfig {@Beanpublic DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");}@Beanpublic Queue errorQueue(){return new Queue("error.queue", true);}@Beanpublic Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");}@Beanpublic MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");}
}

6. 总结

如何确保RabbitMQ消息的可靠性?

  • 开启生产者确认机制,确保生产者的消息能到达队列
  • 开启持久化功能,确保消息未消费前在队列中不会丢失
  • 开启消费者确认机制为auto,由spring确认消息处理成功后完成ack
  • 开启消费者失败重试机制,并设置MessageRecoverer,多次重试失败后将消息投递到异常交换机,交由人工处理

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

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

相关文章

Leetcode 21-25题

合并两个有序链表 将两个升序链表合并为一个新的升序链表。 用两个指针指向两个链表的表头,然后每次比较一下哪个值小,将较小的节点接到答案后面即可。 ListNode* mergeTwoLists(ListNode* list1, ListNode* list2) {auto dummy new ListNode(), p du…

EI级 | Matlab实现TCN-GRU-MATT、TCN-GRU、TCN、GRU多变量时间序列预测对比

EI级 | Matlab实现TCN-GRU-MATT、TCN-GRU、TCN、GRU多变量时间序列预测对比 目录 EI级 | Matlab实现TCN-GRU-MATT、TCN-GRU、TCN、GRU多变量时间序列预测对比预测效果基本介绍程序设计参考资料 预测效果 基本介绍 【EI级】Matlab实现TCN-GRU-MATT、TCN-GRU、TCN、GRU多变量时间…

云原生概念

云原生是一条使用户能: 1.低运维、 2.敏捷的、 3.以可扩展、可复制的方式, 最大化的利用”云“的能力、发挥”云“的价值的最 佳路径 云原生,是一条最佳路径或实践 参考:https://edu.aliyun.com/course/314164/lesson/7815

JavaScript 中实现常见数据结构:栈、队列与树

文章目录 JavaScript 中实现常见数据结构:栈、队列与树引言一、栈(Stack)二、队列(Queue)三、树(Tree)结语与祝福代码 JavaScript 中实现常见数据结构:栈、队列与树 引言 在前端开…

【已解决】引发的异常: 0xC0000005: 读取位置 0xFFFFFFFFFFFFFFFF 时发生访问冲突。

这种问题产生一般都会手足无措,包括笔者,但是不要慌,这种问题一般都是内存泄漏引起的。例如读者要访问一个已经被析构或者释放的变量,当然访问不了,导致存在问题。这时候读者应该从哪里产生内存泄漏这方面进行考虑&…

【VSCode】设置 一键生成vue模板 的快捷入口

问题 每次写一个组件的时候,都需要去手敲默认结构或者是复制粘贴,十分的麻烦! 解决办法 文件 > 首选项 > 用户代码片段 > vue.json 配置vue模板 其中prefix是用来触发代码段的内容,即模版的快捷入口;body里…

【SpringBoot2】Spring容器的本质就是两个HashMap

做为Java程序员,我们就是Spring的用户,Spring的影子在我们的代码里几乎是无处不在,那Spring到底是什么,带给我们什么,如何工作的呢?这个简单问题貌似又不太好讲。例如如果要问一下什么是Spring框架, 可以找到大量的解答,基本内容如下: Spring是一个开源的应用程序框架…

【Funny Game】 吃豆人

目录 【Funny Game】 吃豆人 吃豆人 文章所属专区 Funny Game 吃豆人 吃豆人,这款经典游戏如今依旧魅力四射。玩家需操控小精灵,在迷宫内吞噬所有豆子,同时避开狡猾的鬼怪。当吃完所有豆子后,便可消灭鬼怪,赢得胜利。…

springboot197基于springboot的毕业设计系统的开发

简介 【毕设源码推荐 javaweb 项目】基于springbootvue 的毕业设计系统的开发 适用于计算机类毕业设计,课程设计参考与学习用途。仅供学习参考, 不得用于商业或者非法用途,否则,一切后果请用户自负。 看运行截图看 第五章 第四章 …

[AIGC] 利用 ChatGpt 深入理解 Java 虚拟机(JVM)的内存分布

深入理解 Java 虚拟机(JVM)的内存分布 Java 虚拟机(JVM)是 Java 编程语言的核心运行环境,它负责解释和执行 Java 字节码。在 JVM 中,内存被划分为几个不同的区域,每个区域都有特定的用途。了解…

stack的用法

介绍 翻译为栈&#xff0c;是一个后进先出的容器。 访问 只能通过top()函数来访问栈顶元素 常用函数 1.push(x):将x入队 2.top():获得队首元素 3.empty():判断是否为空 4.size():返回队内元素个数 #include <iostream> #include <stack>//引用stack头文件 us…

c++作业

Shell中的函数&#xff08;先调用后使用的原则&#xff09;&#xff08;没有申明&#xff09; &#xff08;Function&#xff09; 函数名&#xff08;有没有参数根据调用格式&#xff09;&#xff08;不能写任何内容&#xff09; { 函数体 Return 返回值 } 函数名 ----》…

Go语言中的加密艺术:深入解析crypto/subtle库

Go语言中的加密艺术&#xff1a;深入解析crypto/subtle库 引言crypto/subtle库概览ConstantTimeCompare函数深入解析ConstantTimeSelect函数应用详解ConstantTimeLessOrEq函数实践指南安全编程实践性能优化与最佳实践与其他加密库的比较总结 引言 在当今快速发展的互联网时代&…

作为一个程序员,最少要看过这几部电影吧?

计算机专业必看的几部电影 计算机专业必看的几部电影&#xff0c;就像一场精彩的编程盛宴&#xff01;《黑客帝国》让你穿越虚拟世界&#xff0c;感受高科技的魅力&#xff1b;《社交网络》揭示了互联网巨头的创业之路&#xff0c;《源代码》带你穿越时间解救世界&#xff0c;…

SG5032EEN晶体振荡器SPXO

5G将使通信流量呈指数级增长&#xff0c;5G通信网络需要高速和宽带&#xff0c;同时将噪声水平保持在最低水平&#xff0c;这可以通过通信设备的高频低抖动参考时钟来实现&#xff0c;使用上述晶体振荡器SPXO&#xff0c;客户可以输入一个具有极低相位抖动和功率的高频参考时钟…

P5194 Scales S——dfs(前缀和剪枝)

传送门https://www.luogu.com.cn/problem/P5194手切第一遍&#xff0c;没想到可以用前缀和剪枝&#xff0c;并且是从小到大搜了&#xff0c;不出意外40分TLEWA 第二遍&#xff1a;代码及思路 // Problem: // P5194 [USACO05DEC] Scales S // // Contest: Luogu // UR…

施美药业回复监管“二十问”:业务信披存疑,最大客户为关联方

近日&#xff0c;江西施美药业股份有限公司&#xff08;下称“施美药业”&#xff09;更新在深交所主板上市的申请审核动态&#xff0c;对深交所于2023年7月24日发布的第一轮审核问询函进行了回复。目前&#xff0c;深交所已发出第二轮审核问询函。 具体而言&#xff0c;施美药…

1、安装docker

$ sudo apt-get update安装 apt 依赖包&#xff0c;用于通过HTTPS来获取仓库: $ sudo apt-get install \apt-transport-https \ca-certificates \curl \gnupg-agent \software-properties-common添加 Docker 的官方 GPG 密钥&#xff1a; $ curl -fsSL https://mirrors.ustc.…

java:Java中的逻辑控制

顺序结构 在Java中&#xff0c;顺序结构是指代码按照从上到下的顺序依次执行。以下是一个简单的Java程序示例&#xff0c;展示了顺序结构的使用&#xff1a; public class SequenceStructureExample {public static void main(String[] args) {// 声明和初始化变量int num1 10…

cookie+session和token

总结说在前面&#xff1a; session&#xff1a;起源于服务端&#xff0c;保存在服务端&#xff08;服务器或者数据库&#xff09;&#xff0c;通过cookie传递给用户&#xff0c;用户每一次发送HTTP请求的时候&#xff0c;通过验证cookie中的session-id来验证用户身份。jwt(jso…