当前位置: 首页 > news >正文

【RabbitMQ】一文详解消息可靠性

目录:

1.前言

2.生产者

3.数据持久化

4.消费者

5.死信队列

1.前言

RabbitMQ 是一款高性能、高可靠性的消息中间件,广泛应用于分布式系统中。它允许系统中的各个模块进行异步通信,提供了高度的灵活性和可伸缩性。然而,这种通信模式也带来了一些挑战,其中最重要的之一是确保消息的可靠性

影响消息可靠性的因素主要有以下几点:

  • 发送消息时连接RabbitMQ失败
  • 发送时丢失:
    • 生产者发送的消息未送达交换机;
    • 消息到达交换机后未到达队列;
  • MQ 宕机,队列中的消息会丢失;
  • 消费者接收到消息后未消费就宕机了。

2.生产者

2.1.生产者重连机制

生产者发送消息时,出现了网络故障,导致与MQ的连接中断。为了解决这个问题,RabbitMQ提供的消息发送时的重连机制。即:当RabbitTemplate与MQ连接超时后,多次重试。

在生产者yml文件添加配置开启重连机制

spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled: true # 开启超时重试机制initial-interval: 1000ms # 失败后的初始等待时间multiplier: 1 # 失败后下次的等待时长倍数,下次等待时长 = initial-interval * multipliermax-attempts: 3 # 最大重试次数

当网络不稳定的时候,利用重试机制可以有效提高消息发送的成功率。但是RabbitMQ提供的重试机制是阻塞式的重试。 如果对于业务性能有要求,建议禁用重试机制。如果一定要使用,就需要合理配置等待时长和重试次数,或者使用异步线程来执行发送消息的代码

2.2.生产者确认机制

RabbitMQ的生产者确认机制(Publisher Confirm)是一种确保消息从生产者发送到MQ过程中不丢失的机制。当消息发送到 RabbitMQ 后,系统会返回一个结果给消息的发送者,表明消息的处理状态。这个结果有两种可能的值:

返回结果有两种方式:

  • publisher-confirm(发送者确认)
    • 消息成功投递到交换机,返回ACK。
    • 消息未投递到交换机,返回NACK。(可能是由于网络波动未能连接到RabbitMQ,可利用生产者重连机制解决)
  • publisher-return(发送者回执)
    • 消息投递到交换机了,但是没有路由到队列。返回ACK和路由失败原因。(这种问题一般是因为路由键设置错误,可以人为规避)

通过这种机制,生产者在发送消息后获取返回的回执结果,从而采取对应的策略,如消息重发或记录失败信息。

3.数据持久化

3.1.配置持久化

在默认情况下,RabbitMQ会将接收到的信息保存在内存中以降低消息收发的延迟。这样会导致两个问题

  1. RabbitMQ宕机,存在内存中的消息会丢失。
  2. 内存空间有限,当消费者故障或处理过慢时,会导致消息积压,引发MQ阻塞。

为了提升性能,默认情况下MQ的数据都是在内存存储的临时数据,重启后就会消失。RabbitMQ可以通过配置数据持久化,从而将消息保存在磁盘,包括:

  • 交换机持久化(确保RabbitMQ重启后交换机仍然存在)
  • 队列持久化(确保RabbitMQ重启后队列仍然存在)
  • 消息持久化(确保RabbitMQ重启后队列中的消息仍然存在)

由于Spring会在创建队列时默认将交换机和队列设置为持久化,发送消息时也默认指定消息为持久化消息,因此不需要额外配置。

// 将消息指定为持久化消息
Message message = MessageBuilder.withBody("hello".getBytes(standardcharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
// 给队列发送消息
rabbitTemplate.convertAndSend("simple.queue", message);

3.2.惰性队列

从RabbitMQ的3.6.0版本开始,就增加了Lazy Queue的概念,也就是惰性队列

在3.12版本后,所有队列都是Lazy Queue模式,无法更改。

惰性队列的特点如下:

  • 接收到消息后直接存入磁盘而非内存(内存中只保留最近的消息,默认2048条)

  • 消费者要消费消息时才会从磁盘中读取并加载到内存

  • 支持数百万条的消息存储

对于低于3.12版本的情况,可以使用注解的arguments来指定

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "grade.queue", durable = "true"),exchange = @Exchange(name = "intel.topic", type = ExchangeTypes.TOPIC),key = "intel.grade",arguments = @Argument(name = "x-queue-mode", value = "lazy")))

3.3.为什么需要数据持久化?

数据持久化在 RabbitMQ 中有以下重要作用:

队列和交换机的持久化:

  • 防止重启后丢失:将队列和交换机设置为持久化,可以防止 RabbitMQ 服务器重启后丢失这些队列和交换机,确保它们的存在和绑定关系保持不变。

消息的持久化:

  1. 安全性
    • 防止数据丢失:消息持久化后,可以防止 RabbitMQ 服务器重启或宕机时数据丢失,方便数据恢复,保证消息的可靠性和耐久性。
  2. 性能
    • 内存管理:未持久化的临时消息默认存储在内存中。内存空间有限,大量消息涌入时会导致内存占满,系统需要进行 page out 操作将消息写入磁盘。频繁的 page out 操作会严重影响性能。
    • 预防内存溢出:通过持久化消息,可以缓解内存压力,防止因内存溢出导致的系统性能问题和崩溃。

4.消费者

4.1.消费者确认机制

为了确认消费者是否正确处理了消息,RabbitMQ提供了消费者确认机制。当消费者处理消息后,会返回回执信息给RabbitMQ。回执有三种值:

  • ack:消息处理成功,RabbitMQ从队列中删除消息。
  • nack:消息处理失败,RabbitMQ需要再次投递消息。
  • reject:消息处理失败并拒绝该消息,RabbitMQ从队列中删除消息。

在SpringBoot项目中,我们可以通过配置文件选择回执信息的处理方式,一共有三种处理方式:

  • none:不处理。RabbitMQ 假定消费者获取消息后会一定会成功处理,因此消息投递后立即返回ack,将消息从队列中删除。

  • manual:手动模式。需要在业务代码结束后,调用SpringAMQP提供的API发送ackreject,存在代码侵入问题,但比较灵活。

  • auto:自动模式。SpringAMQP利用AOP对我们的消息处理逻辑进行了环绕增强,返回结果如下:

    • 如果消费者正常处理消息,自动返回ack并删除队列的消息。

    • 如果消费者消息处理失败,自动返回nack并重新向消费者投递消息。

    • 如果消息校验异常,自动返回reject并删除队列中的消息。

注意: 手动模式返回回执消息时通常需要显式指定requeue参数,当requeue=true时,表明消息需要重新入队;当requeue=false时,RabbitMQ将从队列删除消息。

spring:rabbitmq:listener:simple:prefetch: 1acknowledge-mode: auto # none,关闭ack;manual,手动ack;auto,自动 ack

4.2.消息失败重试机制

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

可以通过设置yml文件开启失败重试机制,在消息异常时利用本地重试,而不是无限制的进行requeue操作。

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

4.3.消息失败处理策略

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

  • RejectAndDontRequeueRecoverer:重试次数耗尽后,直接reject,丢弃消息,这是默认采取的方式;
  • ImmediateRequeueMessageRecoverer:重试次数耗尽后,返回nack,消息重新入队;
  • RepublishMessageRecoverer:重试耗尽后,将失败消息投递到指定的交换机。

5.死信队列

尽管通过以上设置可以确保消息在生产者、消息队列和消费者之间的传递过程中不会丢失,但在某些情况下,消费者仍可能无法成功处理消息(如消息重试次数耗尽后仍无法被消费)。这时候,我们需要一个机制来妥善处理这些无法被正常消费的消息。死信队列便是用于解决这一问题的兜底机制。

5.1.死信

当一个队列中的消息满足下列情况之一时,可以成为死信(dead letter):

  • 消息被拒绝: 当消费者明确拒绝一个消息并且设置不再重新入队(requeue=false)时,这个消息会被标记为死信。
  • 消息过期: 每个消息或队列可以设置一个TTL(Time-To-Live),即消息的存活时间。如果消息在队列中停留的时间超过了这个TTL,消息会被认为过期,并被转移到死信队列。
  • 队列达到最大长度: 如果队列设置了最大长度并且达到了这个限制,那么新进入的消息会被转移到死信队列中。

5.2.创建死信队列

5.2.1.创建死信交换机和死信队列

正常使用注解,创建交换机和队列即可

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "dead.queue", durable = "true",arguments = @Argument(name = "x-queue-mode", value = "lazy")),exchange = @Exchange(name = "dead.exchange", type = ExchangeTypes.TOPIC),key = "dead.key"
))
public void deadLetterQueue(String msg) {System.out.println("您的消息已经死亡:" + msg);
}
5.2.2.绑定死信交换机

如果队列通过dead-letter-exchange属性指定了一个交换机,那么该队列中的死信就会投递到这个交换机中。这个交换机称为死信交换机(Dead Letter Exchange,简称DLX)

可以通过@Argument注解指定死信交互机和路由键,如下。

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "simple.queue", durable = "true",arguments = {@Argument(name = "x-queue-mode", value = "lazy"),@Argument(name = "x-dead-letter-exchange", value = "dead.exchange"),@Argument(name = "x-dead-letter-routing-key", value = "dead.key")}),exchange = @Exchange(name = "simple.topic",type = ExchangeTypes.TOPIC),key = "simple.key"))

相关文章:

  • 北京网站建设多少钱?
  • 辽宁网页制作哪家好_网站建设
  • 高端品牌网站建设_汉中网站制作
  • 基于springboot和mybatis的RealWorld后端项目实战三之添加swagger
  • HarmonyOS NEXT学习——@BuilderParam装饰器
  • golang mux组件兼容转移url
  • 【Linux】安装PHP扩展-redis
  • 【B树、B-树、B+、B*树】
  • 框架设计MVC
  • 暴力风扇无刷电机驱动方案pcba设计研发
  • 阿里云短信PHP集成api类
  • 【gpt】学习笔记:ChatGPT 用于 3D 内容生成搭配nvidia deepsearch
  • stm32使用通用定时器生成pwm
  • jenkins系列-07.轻易级jpom安装
  • FPGA CFGBVS 管脚接法
  • 彻底改变时尚:使用 GAN 实现 AI 的未来
  • 【自撰写】【国际象棋入门】第11课 对局实例分析(一)
  • LabVIEW红外热波图像缺陷检
  • 分享一款快速APP功能测试工具
  • [rust! #004] [译] Rust 的内置 Traits, 使用场景, 方式, 和原因
  • 「前端」从UglifyJSPlugin强制开启css压缩探究webpack插件运行机制
  • 【MySQL经典案例分析】 Waiting for table metadata lock
  • 【前端学习】-粗谈选择器
  • 【跃迁之路】【463天】刻意练习系列222(2018.05.14)
  • docker-consul
  • download使用浅析
  • Electron入门介绍
  • github从入门到放弃(1)
  • JSDuck 与 AngularJS 融合技巧
  • SQL 难点解决:记录的引用
  • Vue小说阅读器(仿追书神器)
  • Webpack4 学习笔记 - 01:webpack的安装和简单配置
  • weex踩坑之旅第一弹 ~ 搭建具有入口文件的weex脚手架
  • 创建一种深思熟虑的文化
  • 基于axios的vue插件,让http请求更简单
  • 漫谈开发设计中的一些“原则”及“设计哲学”
  • 猫头鹰的深夜翻译:Java 2D Graphics, 简单的仿射变换
  • 前端面试总结(at, md)
  • 小程序测试方案初探
  • 一些关于Rust在2019年的思考
  • 优化 Vue 项目编译文件大小
  • 中国人寿如何基于容器搭建金融PaaS云平台
  • puppet连载22:define用法
  • SAP CRM里Lead通过工作流自动创建Opportunity的原理讲解 ...
  • ​【C语言】长篇详解,字符系列篇3-----strstr,strtok,strerror字符串函数的使用【图文详解​】
  • ‌前端列表展示1000条大量数据时,后端通常需要进行一定的处理。‌
  • !!java web学习笔记(一到五)
  • (0)Nginx 功能特性
  • (delphi11最新学习资料) Object Pascal 学习笔记---第2章第五节(日期和时间)
  • (k8s中)docker netty OOM问题记录
  • (阿里巴巴 dubbo,有数据库,可执行 )dubbo zookeeper spring demo
  • (论文阅读22/100)Learning a Deep Compact Image Representation for Visual Tracking
  • (强烈推荐)移动端音视频从零到上手(上)
  • (原創) 人會胖會瘦,都是自我要求的結果 (日記)
  • (转)h264中avc和flv数据的解析
  • (转载)PyTorch代码规范最佳实践和样式指南
  • .360、.halo勒索病毒的最新威胁:如何恢复您的数据?
  • .Net 8.0 新的变化