RabbitMQ问题总结


:::info
使用场景

  • 异步发送(验证码、短信、邮件。。。)
  • MySQL 和 Redis、ES 之间的数据同步
  • 分布式事务
  • 削峰填谷

  • :::

如何保证消息不丢失


上图是消息正常发送的一个过程,那在哪个环节中消息容易丢失?在哪一个环节都可能丢失

  • 生产者宕机,消息就可能到达不了交换机,或者消息未到达队列
  • 消息发送成功后,消费者还没有消费前,MQ 宕机,就有可能导致队列中消息丢失
  • 消费者宕机,导致消费者未接收到消息

生产者,队列,消费者三个层面都有可能导致消息丢失,所以保证消息不丢失需要从这三个层面解决

生产者确认机制

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

如果消息发送成功,就会返回 publish-confirm ack,如果发送到交换机失败,就会返回 publish-confirm nack,如果发送到队列失败,就会返回 publish-return ack。
消息失败之后如何处理呢?

  • 回调方法即时重发
  • 记录日志
  • 保存到数据库然后定时重发,成功发送后即刻删除表中的数据

消息持久化

MQ 默认是内存存储消息,开启持久化功能可以确保缓存在 MQ 中的消息不丢失。

  1. 交换机持久化
@Bean
public DirectExchange simpleExchange(){// 三个参数:交换机名称、是否持久化、当没有queue与其绑定时是否自动删除return new DirectExchange("simple.direct",true,false);
}
  1. 队列持久化
@Bean
public Queue simpleQueue(){// 使用QueueBuilder构建队列,durable就是持久的return QueueBuilder.durable("simple.queue").build();
}
  1. 消息持久化,SpringAMQP 中的消息默认是持久的,可以通过 MessageProperties 中的 DeliveryMode 来指定
Message msg = MessageBuilder.withBody(message.getBytes(StandardCharsets.UTF_8)) // 消息体.setDeliveryMode(MessageDeliveryMode.PERSISTENE) // 持久化.build();

消费者确认

RabbitMQ 支持消费者确认机制,即消费者处理消息后可以向 MQ 发送 ack 回执,MQ 收到 ack 回执后才会删除消息。
SpringAMQP 则允许配置三种确认模式:

  • manual:手动 ack,需要在业务代码结束后,调用 api 发送 ack。
  • auto:自动 ack,由 spring 监测 listener 代码是否出现异常,没有异常则返回 ack;抛出异常则返回 nack。
  • none:关闭 ack,MQ 假定消费者获取消息后会成功处理,因此消息投递后立即被删除

我们可以利用 Spring 的 retry 机制,在消费者出现异常时利用本地重试,设置重试次数,当次数达到了以后,如果消息依然失败,将消息投递到异常交换机,交由人工处理。

RabbitMQ 如何保证消息不丢失?

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

消息的重复消费问题如何解决的

为什么会出现重复消费的问题?

  • 网络抖动
  • 消费者挂了


消费者已经处理完消息,还没来得及给 MQ 发送确认,这时网络发生了抖动或者消费者挂了,等网络恢复之后或者消费者重启之后,因为队列没有收到确认,所以消息还在 MQ 中,因为我们设置了重试机制,消费者就会重新消费消息。
解决方案

  • 每条消息设置一个唯一的标识 id,消费者收到消息后去业务 id 是否存在
  • 幂等方案:分布式锁、数据库锁(悲观锁、乐观锁)

RabbitMQ 死信交换机(RabbitMQ 延迟队列有了解过嘛)

  • 延迟队列:进入队列的消息会被延迟消费的队列
  • 场景:超时订单、限时优惠、定时发布

延迟队列=死信交换机+TTL(消息的生存时间)

死信交换机

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

  • 消费者使用 basic.reject 或 basic.nack 声明消费失败,并且消息的 requeue 参数为 false
  • 消息是一个过期消息,超时无人消费
  • 要投递的队列消息堆积满了,最早的消息可能成为死信

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

@Bean
public Queue ttlQueue(){return QueueBuilder.durable("simple.queue")  // 指定队列名称,并持久化.ttl(10000)  // 设置队列的超时时间,10秒.deadLetterExchange("dl.direct")  // 指定死信交换机.build();
}

TTL

TTL,也就是 Time-To-Live。如果一个队列中的消息 TTL 结束仍未消费,则会变成死信,ttl 超时分为两种情况:

  • 消息所在的队列设置了存活时间
  • 消息本身设置了存活时间


哪个 ttl 短以哪个为准。

// 创建消息
Message message = MessageBuilder.withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8)).setExpiration("5000").build();
// 消息ID,需要封装到CorrelationData中
CorrelationData correlationData = new correlationData(UUID.randomUUID().toString());
// 发送消息
rabbitTemplate.convertAndSend("ttl.direct","ttl",message,correlationData);

延迟队列插件

实现延迟队列,还可以使用官方提供的插件。DelayExchange 插件,需要安装在 RabbitMQ 中
RabbitMQ 有一个官方的插件社区,地址为:https://www.rabbitmq.com/community-plugins.htmlimage.png

@RabbitListener(bindings = @QueueBinding(value = @Queue(name ="delay.queue", durable="true"),exchange = @Exchange(name="delay.direct",delayed="true")key="delay"
))
public void listenDelayedQueue(String msg){log.info("接收到 delay.queue的延迟消息:{}",msg);
}
// 创建消息
Message message = MessageBuilder.withBody("hello,delayed message".getBytes(StandardCharsets.UTF_8)).setHeader("x-delay",10000).build();
// 消息ID,需要封装到CorrelationData中
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
// 发送消息
rabbitTemplate.convertAndSend("delay.direct","delay",message,correlationData);

RabbitMQ 死信交换机?(RabbitMQ 延迟队列有了解过嘛)

  • 我们当时一个业务使用到了延迟队列(超时订单、限时优惠、定时发布…)
  • 其中延迟队列就用到了死信交换机和 TTL 实现的
  • 消息超时未消费就会变成死信(死信的其他情况:拒绝被消费,队列满了)

延迟队列插件实现延迟队列 DelayExchange

  • 声明一个交换机,添加 delayed 属性为 true
  • 发送消息时,添加 x-delay 头,值为超时时间

如果有 100 万消息堆积在 MQ,如何解决(消息堆积怎么解决)

当生产者发送消息的速度超过了消费者处理消息的速度,就会导致队列中的消息堆积,直到队列存储消息达到上限。之后发送的消息就会成为死信,可能会被丢弃,这就是消息堆积问题。
解决消息堆积有三种思路:

  • 增加更多消费者,提高消费速度
  • 在消费者内开启线程池加快消息处理速度
  • 扩大队列容积,提高堆积上限

惰性队列

惰性队列的特征如下:

  • 接收到消息后直接存入磁盘而非内存
  • 消费者要消费消息时才会从磁盘中读取并加载到内存
  • 支持百万条的消息存储
@Bean
public Queue lazyQueue(){return QueueBuilder.durable("lazy.queue").lazy()  // 开启x-queue-mode为lazy.build();
}
// 或者
@RabbitListener(queuesToDeclare = @Queue(name="lazy.queue",durable="true",arguments=@Argument(name="x-queue-mode",value="lazy")
))
public void listenLazyQueue(String msg){log.info("接收到lazy.queue的消息:{}",msg);
}

如果有 100 万消息堆积在 MQ,如何解决?
解决消息堆积有三种思路:

  • 增加更多的消费者,提高消费速度
  • 在消费者内开启线程池加快消息处理速度
  • 扩大队列容积,提高堆积上限,采用惰性队列
    • 在声明队列的时候可以设置属性 x-queue-mode 为 lazy,即为惰性队列
    • 基于磁盘存储,消息上限高
    • 性能比较稳定,但基于磁盘存储,受限与磁盘 IO,时效性会降低

RabbitMQ 高可用机制

  • 在生产环境下,使用集群来保证高可用性
  • 普通集群、镜像集群、仲裁队列

普通集群

普通集群,或者叫标准集群(classic cluster),具备下列特征:

  • 会在集群的各个节点共享部分数据,包括:交换机、队列元信息。不包含队列中的消息。
  • 当访问集群某节点时,如果队列不在该节点,会从数据所在节点传递到当前节点并返回。
  • 队列所在节点宕机,队列中的消息就会丢失。

镜像集群

镜像集群:本质是主从模式,具备下面的特征:

  • 交换机、队列、队列中的消息会在各个 mq 的镜像节点之间同步备份。
  • 创建队列的节点被称为该队列的主节点,备份到其他节点叫做该队列的镜像节点。
  • 一个队列的主节点可能是另一个队列的镜像节点。
  • 所有操作都是在主节点完成,然后同步给镜像节点。
  • 主宕机后,镜像节点会替代成新的主。

仲裁队列

仲裁队列:仲裁队列是 3.8 版本以后才有的新功能,用来替代镜像队列,具备下列特征:

  • 与镜像队列一样,都是主从模式,支持主从数据同步
  • 使用非常简单,没有复杂的配置
  • 主从同步基于 Raft 协议,强一致
@Bean
public Queue quorumQueue(){return QueueBuilder.durable("quorum.queue")  // 持久化.quorum()  //仲裁队列.build();
}

RabbitMQ 的高可用机制有了解过嘛

  • 在生产环境下,我们当时采用的镜像模式搭建的集群,共有 3 个节点。
  • 镜像队列结构是一主多从(从就是镜像),所有操作都是主节点完成,然后同步给镜像节点。
  • 主节点宕机后,镜像节点会替代成新的主(如果在主从同步完成前,主就已经宕机,可能出现数据丢失)

出现数据丢失怎么解决?
我们可以采用仲裁队列,与镜像队列一样,都是主从模式,支持主从数据同步,主从同步基于 Raft 协议,强一致。并且使用起来也非常简单,不需要额外的配置,在声明队列的时候只要指定这个是仲裁队列即可。

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

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

相关文章

opencv#32 可分离滤波

滤波的可分离性 就是将一个线性滤波变成多个线性滤波,这里面具体所指的是变成x方向的线性滤波和y方向的线性滤波。无论先做x方向的滤波还是y方向滤波,两者的叠加结果是一致的,这个性质取决于滤波操作是并行的,也就是每一个图像在滤…

vue3+echarts绘制某省区县地图

vue3echarts绘制某省区县地图 工作中经常需要画各种各样的图,echarts是使用最多的工具,接近春节,想把之前画的echarts图做一个整合,方便同事和自己随时使用,因此用vue3专门写了个web项目,考虑之后不断完善…

基于SpringBoot实现策略模式提供系统接口扩展能力

相信我们对策略模式都有耳闻,但是可能不知道它在项目中具体能有什么作用,我们需要在什么场景下才能去尽可能得去使用策略模式。 这里我简单的列出一个我之前在公司做的一个需求:跟第三方oa系统对接接口,对方需要回调我们当前系统…

R12.2 EBS 修改 APPS 密码 详细步骤

目录 前言准备修改步骤1.关闭应用层2.FNDCPASS 修改密码3. 运行 autoconfig4.单独启动 webLogic 服务5.登录weblogic,更新apps密码6.启动应用层7.验证 结尾 前言 本文的目的是修改 apps 密码,主要参考官方文档 metalink 1674462.1,请注意本文…

代码随想录算法训练营Day|层序遍历,翻转二叉树,对称二叉树

层序遍历 层序遍历一个二叉树,就是从左往右一层一层的遍历二叉树,这种遍历方式需要借用一个辅助数据结构即队列来实现,队列先进先出,符合一层一层遍历的逻辑,而用栈先进后出适合模拟深度优先遍历也就是递归的逻辑。而…

Task04:DDPG、TD3算法

本篇博客是本人参加Datawhale组队学习第四次任务的笔记 【教程地址】https://github.com/datawhalechina/joyrl-book 【强化学习库JoyRL】https://github.com/datawhalechina/joyrl/tree/main 【JoyRL开发周报】 https://datawhale.feishu.cn/docx/OM8fdsNl0o5omoxB5nXcyzsInGe…

Delphi 7 IdHTTP POST 中文乱码得解决

WEB后台使用UTF-8进行编码,由于D7默认是ansiString,直接提交到后台会使中文乱码。 解决方法: 1.先把AnsiString转WideString 2.通过System单元中的ansitoUTF8()函数进行转换之后再提交就可以了。 代码示例: procedure postDe…

华为三层交换机之基本操作

Telnet简介 Telnet是一个应用层协议,可以在Internet上或局域网上使用。它提供了基于文本的远程终端接口,允许用户在本地计算机上登录到远程计算机,然后像在本地计算机上一样使用远程计算机的资源。Telnet客户端和服务器之间的通信是通过Telnet协议进行的…

OpenAI 降低价格并修复拒绝工作的“懒惰”GPT-4,另外ChatGPT 新增了两个小功能

OpenAI降低了GPT-3.5 Turbo模型的API访问价格,输入和输出价格分别降低了50%和25%。这对于使用API进行文本密集型应用程序的用户来说是一个好消息。 OpenAI官网:OpenAI AIGC专区:aigc 教程专区:AI绘画,AI视频&#x…

Vite+Electron快速构建一个VUE3桌面应用(二)——动态模块热重载

一. 简介 在上一篇文章ViteElectron快速构建一个VUE3桌面应用中,我们了解了如何使用Vite和Electron来快速构建一个Vue3桌面应用。但是,之前构建的应用仅仅是一个简单的版本。在开发过程中,为了更好的开发体验,在开发electron的时…

博弈论(牛客练习赛)

思路:我们考虑小念赢 1、如果n>1并且p0,小念可以连续取两次,相当于小念有挂,可以从必败态转为必胜态,必赢。 2、如果n>1并且m>n-1,小念第一次取n-1个,小念必赢。 代码: …

【C语言】(1)初识C语言

什么是C语言 C语言是一种广泛应用的计算机编程语言,它具有强大的功能和灵活性,使其成为系统编程和底层开发的首选语言。C语言的设计简洁、高效,且不依赖于特定的硬件或系统,因此在各种计算平台上都能稳定运行。 C语言的特点 高…

C++ 之LeetCode刷题记录(十九)

😄😊😆😃😄😊😆😃 开始cpp刷题之旅。 依旧是追求耗时0s的一天。 108. 将有序数组转换为二叉搜索树 给你一个整数数组 nums ,其中元素已经按 升序 排列,请你…

针对于vue element-plus组件的el-date-picker日期区间组件的日期格式问题以及如何进行区间判断

<template><el-date-picker v-model"value1" type"daterange" range-separator"To" start-placeholder"开始日期" end-placeholder"结束日期" :size"size" change"sarend" /> </templat…

速盾:海外网站CDN加速建站有哪些优势?

随着互联网的发展&#xff0c;海外网站的建设和访问需求也日益增加。而为了提高海外网站的访问速度和用户体验&#xff0c;CDN加速技术成为了一种必不可少的解决方案。本文将介绍海外网站CDN加速建站的优势&#xff0c;并回答一些相关问题。 一、提高网站访问速度 海外网站部署…

C#学习笔记_数组

数组是一个存储相同类型元素的固定大小的顺序集合。数组是用来存储数据的集合&#xff0c;通常认为数组是一个同一类型变量的集合。 声明数组 一、一维数组 一维数组声明语法如下&#xff1a; datatype[] arrayName; 其中&#xff0c;datatype为数据类型&#xff0c;array…

844.比较含退格的字符串(力扣LeetCode)

844.比较含退格的字符串 题目描述 给定 s 和 t 两个字符串&#xff0c;当它们分别被输入到空白的文本编辑器后&#xff0c;如果两者相等&#xff0c;返回 true 。# 代表退格字符。 注意&#xff1a;如果对空文本输入退格字符&#xff0c;文本继续为空。 示例 1&#xff1a;…

泽众云真机-机型集中化运维方案升级全面完成

2024年元月份&#xff0c;泽众云真机运维团队&#xff0c;经过几个月软硬件多轮安装调试&#xff0c;机型集中化运维方案升级全面完成。解决了云真机的机型集中化运维难题&#xff0c;方便了运营人员手机管理。 具体如下&#xff1a; 1、集中化运维&#xff0c;如服务器、PC、…

中级ccnp多久可以考下来?ccnp 课件哪里有?

思科中级CCNP考试要求考生具备一定的网络基础知识&#xff0c;包括IP地址规划、VLAN配置、STP协议等。然而&#xff0c;对于许多初学者来说&#xff0c;中级CCNP考试的难度和门槛都是一个不小的挑战。那么&#xff0c;究竟需要多久才能考下中级CCNP证书呢?下面将为你详细解读。…

esp32 操作DS1307时钟芯片

电气参数摘要 有VCC供电&#xff0c;IIC活动状态是1.5mA&#xff0c;待机状态200μA&#xff0c;电池电流5nA(MAX50nA&#xff09;无VCC供电的时候&#xff0c;电池电流&#xff0c;300nA&#xff08;时钟运行&#xff09;&#xff0c;10nA&#xff08;时钟停止&#xff09;供…