RabbitMQ常见问题

一、RabbitMQ如何保证消息不丢失?

        
        这是面试时最喜欢问的问题,其实这是个所有MQ的一个共性的问题,大致的解 决思路也是差不多的,但是针对不同的MQ产品会有不同的解决方案。而RabbitMQ 设计之处就是针对企业内部系统之间进行调用设计的,所以他的消息可靠性是比较 高的。

1、哪些环节会有丢消息的可能?

我们考虑一个通用的MQ场景:

其中,1,2,4三个场景都是跨网络的,而跨网络就肯定会有丢消息的可能。然后关于3这个环节,通常MQ存盘时都会先写入操作系统的缓存page cache中,然后再由操作系统异步的将消息写入硬盘。这个中间有个时间差,就可能会造成消息丢失。如果服务挂了,缓存中还没有来得及写入硬盘的消息就会丢失。这也是任何用户态的应用程序无法避免的。对于任何MQ产品,都应该从这四个方面来考虑数据的安全性。那我们看看用RabbitMQ时要如何解决这个问题。

2、RabbitMQ消息零丢失方案


1》生产者保证消息正确发送到RibbitMQ


对于单个数据,可以使用生产者确认机制。通过多次确认的方式,保证生产者的消息能够正确的发送到RabbitMQ中。RabbitMQ的生产者确认机制分为同步确认和异步确认。同步确认主要是通过在生产者端使用Channel.waitForConfirmsOrDie()指定一个等待确认的完成时间。异步确认机制则是通过channel.addConfirmListener(ConfirmCallback var1,ConfirmCallback var2)在生产者端注入两个回调确认函数。第一个函数是在生产者发送消息时调用,第二个函数则是生产者收到Broker的消息确认请求时调用。两个函数需要通过sequenceNumber自行完成消息的前后对应。sequenceNumber的生成方式需要通过channel的序列获取。

int sequenceNumber = channel.getNextPublishSeqNo();
在RabbitMQ中,另外还有一种手动事务的方式,可以保证消息正确发送。

手动事务机制主要有几个关键的方法:

channel.txSelect() 开启事务;

channel.txCommit() 提交事务;

channel.txRollback() 回滚事务;

用这几个方法来进行事务管理。但是这种方式需要手动控制事务逻辑,并且手动事务会对channel产生阻塞,造成吞吐量下降

2》 RabbitMQ消息存盘不丢消息


这个在RabbitMQ中比较好处理,对于Classic经典队列,直接将队列声明成为持久化队列即可。而新增的Quorum队列和Stream队列,都是明显的持久化队列,能更好的保证服务端消息不会丢失。


3》 RabbitMQ 主从消息同步时不丢消息


这涉及到RabbitMQ的集群架构。首先他的普通集群模式,消息是分散存储的,不会主动进行消息同步了,是有可能丢失消息的。而镜像模式集群,数据会主动在集群各个节点当中同步,这时丢失消息的概率不会太高。另外,启用Federation联邦机制,给包含重要消息的队列建立一个远端备份,也是一个不错的选择。


4》 RabbitMQ消费者不丢失消息


RabbitMQ在消费消息时可以指定是自动应答,还是手动应答。如果是自动应答模式,消费者会在完成业务处理后自动进行应答,而如果消费者的业务逻辑抛出异常,RabbitMQ会将消息进行重试,这样是不会丢失消息的,但是有可能会造成消息一直重复消费。将RabbitMQ的应答模式设定为手动应答可以提高消息消费的可靠性。另外这个应答模式在SpringBoot集成案例中,也可以在配置文件中通过属性spring.rabbitmq.listener.simple.acknowledge-mode 进行指定。可以设定AUTO 自动应答; MANUAL 手动应答;NONE 不应答; 其中这个NONE不应答,就是不启动应答机制,RabbitMQ只管往消费者推送消息后,就不再重复推送消息了,相当于RocketMQ的sendoneway, 这样效率更高,但是显然会有丢消息的可能。


最后,任何用户态的应用程序都无法保证绝对的数据安全,所以,备份与恢复的方案也需要考虑到。


二、如何保证消息幂等?


1、RabbitMQ的自动重试功能:


当消费者消费消息处理业务逻辑时,如果抛出异常,或者不向RabbitMQ返回响应,默认情况下,RabbitMQ会无限次数的重复进行消息消费。处理幂等问题,首先要设定RabbitMQ的重试次数。在SpringBoot集成RabbitMQ时,可以在配置文件中指定spring.rabbitmq.listener.simple.retry开头
的一系列属性,来制定重试策略。然后,需要在业务上处理幂等问题。处理幂等问题的关键是要给每个消息一个唯一的标识。在SpringBoot框架集成RabbitMQ后,可以给每个消息指定一个全局唯一的MessageID,在消费者端针对MessageID做幂等性判断。关键代码:

channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope
envelope,
BasicProperties properties, byte[]
body)
throws IOException {
long deliveryTag = envelope.getDeliveryTag();
channel.basicAck(deliveryTag, false);
}
});
channel.basicConsume(queueName, true, myconsumer);


要注意下这里用的message要是org.springframework.amqp.core.Message在原生API当中,也是支持MessageId的。当然,在实际工作中,最好还是能够添加一个具有业务意义的数据作为唯一键会更好,这样能更好的防止重复消费问题对业务的影响。比如,针对订单消息,那就用订单ID来做唯一键。在RabbitMQ中,消息的头部就是一个很好的携带数据的地方。

//发送者指定ID字段
Message message2 =
MessageBuilder.withBody(message.getBytes()).setMessageId(UUID.randomUUID().
toString()).build();
rabbitTemplate.send(message2);
//消费者获取MessageID,自己做幂等性判断
@RabbitListener(queues = "fanout_email_queue")
public void process(Message message) throws Exception {
// 获取消息Id
String messageId = message.getMessageProperties().getMessageId();
...
}// ==== 发送消息时,携带sequenceNumber和orderNo
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.deliveryMode(MessageProperties.PERSISTENT_TEXT_PLAIN.getDeliveryMod
e());
builder.priority(MessageProperties.PERSISTENT_TEXT_PLAIN.getPriority());
//携带消息ID
builder.messageId(""+channel.getNextPublishSeqNo());
Map<String, Object> headers = new HashMap<>();
//携带订单号
headers.put("order", "123");
builder.headers(headers);
channel.basicPublish("", QUEUE_NAME, builder.build(),
message.getBytes("UTF-8"));
// ==== 接收消息时,拿到sequenceNumber
Consumer myconsumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope
envelope,
BasicProperties properties, byte[] body)
throws IOException {
//获取消息ID
System.out.println("messageId:"+properties.getMessageId());
//获取订单ID



三、如何保证消息的顺序?


某些场景下,需要保证消息的消费顺序,例如一个下单过程,需要先完成扣款,然后扣减库存,然后通知快递发货,这个顺序不能乱。如果每个步骤都通过消息进行异步通知的话,这一组消息就必须保证他们的消费顺序是一致的。
在RabbitMQ当中,针对消息顺序的设计其实是比较弱的。唯一比较好的策略就是 单队列+单消息推送。即一组有序消息,只发到一个队列中,利用队列的FIFO特性保证消息在队列内顺序不会乱。但是,显然,这是以极度消耗性能作为代价的,在实际适应过程中,应该尽量避免这种场景。
然后在消费者进行消费时,保证只有一个消费者,同时指定prefetch属性为1,即每次RabbitMQ都只往客户端推送一个消息。像这样:而在多队列情况下,如何保证消息的顺序性,目前使用RabbitMQ的话,还没有比较好的解决方案。在使用时,应该尽量避免这种情况。


四、关于RabbitMQ的数据堆积问题


RabbitMQ一直以来都有一个缺点,就是对于消息堆积问题的处理不好。当RabbitMQ中有大量消息堆积时,整体性能会严重下降。而目前新推出的Quorum队列以及Stream队列,目的就在于解决这个核心问题。但是这两种队列的稳定性和周边生态都还不够完善,因此,在使用RabbitMQ时,还是要非常注意消息堆积的问题。尽量让消息的消费速度和生产速度保持一致。
 

properties.getHeaders().forEach((key,value)->
System.out.println("key: "+key +"; value: "+value));
// (process the message components here ...)
//消息处理完后,进行答复。答复过的消息,服务器就不会再次转发。
//没有答复过的消息,服务器会一直不停转发。
channel.basicAck(deliveryTag, false);
}
};
channel.basicConsume(QUEUE_NAME, false, myconsumer);spring.rabbitmq.listener.simple.prefetch=11


而如果确实出现了消息堆积比较严重的场景,就需要从数据流转的各个环节综合考虑,设计适合的解决方案。


首先在消息生产者端:

对于生产者端,最明显的方式自然是降低消息生产的速度。但是,生产者端产生消息的速度通常是跟业务息息相关的,一般情况下不太好直接优化。但是可以选择尽量多采用批量消息的方式,降低IO频率。


然后在RabbitMQ服务端:
从前面的分享中也能看出,RabbitMQ本身其实也在着力于提高服务端的消息堆积能力。对于消息堆积严重的队列,可以预先添加懒加载机制,或者创建Sharding分片队列,这些措施都有助于优化服务端的消息堆积能力。另外,尝试使用Stream队列,也能很好的提高服务端的消息堆积能力。


接下来在消息消费者端:
要提升消费速度最直接的方式,就是增加消费者数量了。尤其当消费端的服务出现问题,已经有大量消息堆积时。这时,可以尽量多的申请机器,部署消费端应用,争取在最短的时间内消费掉积压的消息。但是这种方式需要注意对其他组件的性能压力。对于单个消费者端,可以通过配置提升消费者端的吞吐量。例如灵活配置这几个参数,能够在一定程度上调整每个消费者实例的吞吐量,减少消息堆积数量。当确实遇到紧急状况,来不及调整消费者端时,可以紧急上线一个消费者组,专门用来将消息快速转录。保存到数据库或者Redis,然后再慢慢进行处理。


五、RabbitMQ的备份与恢复


文档地址: https://www.rabbitmq.com/backup.html
# 单次推送消息数量
spring.rabbitmq.listener.simple.prefetch=1
# 消费者的消费线程数量
spring.rabbitmq.listener.simple.concurrency=5

RabbitMQ有一个data目录会保存分配到该节点上的所有消息。我们的实验环境中,默认是在/var/lib/rabbitmq/mnesia目录下 这个目录里面的备份分为两个部分,一个是元数据(定义结构的数据),一个是消息存储目录。对于元数据,可以在Web管理页面通过json文件直接导出或导入。而对于消息,可以手动进行备份恢复其实对于消息,由于MQ的特性,是不建议进行备份恢复的。而RabbitMQ如果要进行数据备份恢复,也非常简单。

首先,要保证要恢复的RabbitMQ中已经有了全部的元数据,这个可以通过上一步的json文件来恢复。然后,备份过程必须要先停止应用。如果是针对镜像集群,还需要把整个集群全部停止。最后,在RabbitMQ的数据目录中,有按virtual hosts组织的文件夹。你只需要按照虚拟主机,将整个文件夹复制到新的服务中即可。持久化消息和非持久化消息都会一起备份。 我们实验环境的默认目录是/var/lib/rabbitmq/mnesia/rabbit@worker2/msg_stores/vhosts


六、总结


基于MQ的事件驱动机制,给庞大的互联网应用带来了不一样的方向。MQ的异步、解耦、削峰三大功能特点在很多业务场景下都能带来极大的性能提升,在日常工作过程中,应该尝试总结这些设计的思想。虽然MQ的功能,说起来比较简单,但是随着MQ的应用逐渐深化,所需要解决的问题也更深入。对各种细化问题的挖掘程度,很大程度上决定了开发团队能不能真正Hold得住MQ产品。

通常面向互联网的应用场景,更加注重MQ的吞吐量,需要将消息尽快的保存下来,再供后端慢慢消费。

而针对企业内部的应用场景,更加注重MQ的数据安全性,在复杂多变的业务场景下,每一个消息都需要有更加严格的安全保障。

而在当今互联网,Kafka是第一个场景的不二代表,但是他会丢失消息的特性,让kafka的使用场景比较局限。

RabbitMQ作为一个老牌产品,是第二个场景最有力的代表。当然,随着互联网应用不端成熟,也不断有其他更全能的产品冒出来,比如阿里的RocketMQ以及雅虎的Pulsar。但是不管未来MQ领域会是什么样子,RabbitMQ依然是目前企业级最为经典也最为重要的一个产品。他的功能最为全面,周边生态也非常成熟,并且RabbitMQ有庞大的Spring社区支持,本身也在吸收其他产品的各种优点,持续进化,所以未来RabbitMQ的重要性也会更加凸显。
 

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

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

相关文章

递归学习——记忆化搜索

目录 ​编辑 一&#xff0c;概念和效果 二&#xff0c;题目 1.斐波那契数 1.题目 2.题目接口 3.解题思路 2.不同的路径 1.题目 2.题目接口 3.解题思路 3.最长增长子序列 1.题目 2.题目接口 3.解题思路 4.猜数字游戏II 1.题目 2.题目接口 3.解题思路 总结&a…

【SpringMVC】Jrebel 插件实现热部署与文件上传

目录 一、JRebel 1.1 Jrebel介绍 1.2 Jrebel插件下载 1.3 Jrebel服务下载并启动 1.4 在线生成GUID 1.5 JRebel激活 1.6 相关设置 注意❗ 二、文件上传、下载 2.1 导入pom依赖 2.2 配置文件上传解析器 2.3 文件上传表单设置 2.4 文件上传实现 2.5 文件下载实现 2…

微信会员卡开发流程

功能需求&#xff1a; 通过微信第三方平台创建的模板小程序&#xff0c;想要实现用户在小程序支付一定金额后领取会员卡&#xff0c;领取会员卡后可给用户下发一定数量的优惠券&#xff0c;并且实现用户在小程序消费享受商品折扣。 开发流程&#xff1a; 一、了解微信的3个平…

HTTP协议的基本概念与理解!

一、什么是HTTP协议 HTTP&#xff08;超文本传输协议&#xff09;是一个基于请求与响应&#xff0c;无状态的&#xff0c;应用层的协议&#xff0c;常基于TCP/IP协议传输数据&#xff0c;互联网上应用最为广泛的一种网络协议,所有的WWW文件都必须遵守这个标准。设计HTTP的初衷…

【RocketMQ】消息的拉取

在上一讲中&#xff0c;介绍了消息的存储&#xff0c;生产者向Broker发送消息之后&#xff0c;数据会写入到CommitLog中&#xff0c;这一讲&#xff0c;就来看一下消费者是如何从Broker拉取消息的。 RocketMQ消息的消费以组为单位&#xff0c;有两种消费模式&#xff1a; 广播…

实时显示当前文件夹下的文件大小,shell脚本实现

图片来源于网络&#xff0c;如果侵权请联系博主删除&#xff01; 需求&#xff1a; 写一个shell终端命令&#xff0c;实时显示当前文件夹下的文件大小 实现&#xff1a; 您可以使用以下的Shell脚本命令来实时显示当前文件夹下的文件大小&#xff1a; while true; docleardu …

百度飞浆OCR识别表格入门python实践

1. 百度飞桨&#xff08;PaddlePaddle&#xff09; 百度飞桨&#xff08;PaddlePaddle&#xff09;是百度推出的一款深度学习平台&#xff0c;旨在为开发者提供强大的深度学习框架和工具。飞桨提供了包括OCR&#xff08;光学字符识别&#xff09;在内的多种功能&#xff0c;可…

《动手学深度学习 Pytorch版》 4.10 实战Kaggle比赛:预测比赛

4.10.1 下载和缓存数据集 import hashlib import os import tarfile import zipfile import requests#save DATA_HUB dict() DATA_URL http://d2l-data.s3-accelerate.amazonaws.com/def download(name, cache_diros.path.join(.., data)): #save"""下载一个…

【chromium】windows 获取源码到本地

从github的chromium 镜像git clone 到2.5G失败了官方说不能,要去 windows_build_instructions vs2017和19都是32位的 vs2022是x64的 vs2022_install You may also have to set variable vs2022_install to your installation path of Visual Studio 2022,

自定义Dynamics 365实施和发布业务解决方案 3. 开箱即用自定义

在本章中,您将开始开发SBMA会员应用程序。在开发的最初阶段,主要关注开箱即用的定制。在第2章中,我们讨论了如何创建基本解决方案的细节,在本章中,将创建作为解决方案补丁的基本自定义,并展示将解决方案添加到源代码管理和目标环境的步骤。 表单自定义 若要开始表单自定…

宠物行业如何进行软文营销

如今&#xff0c;宠物已经成为了人们生活中不可或缺的一部分&#xff0c;大众对于萌宠的喜爱与日俱增&#xff0c;随着“萌宠经济”升温&#xff0c;越来越多的商机开始出现&#xff0c;伴随着宠物市场竞争的日益激烈&#xff0c;宠物行业的营销光靠硬广告很难吸引受众&#xf…

使用内网端口映射方案,轻松实现U8用友ERP的本地部署异地远程访问——“cpolar内网穿透”

文章目录 前言1. 服务器本机安装U8并调试设置2. 用友U8借助cpolar实现企业远程办公2.1 在被控端电脑上&#xff0c;点击开始菜单栏&#xff0c;打开设置——系统2.2 找到远程桌面2.3 启用远程桌面 3. 安装cpolar内网穿透3.1 注册cpolar账号3.2 下载cpolar客户端 4. 获取远程桌面…

小美的数组操作2---牛客周赛 Round 11

注意给a[ 0 ]赋一个最小值 #include<bits/stdc.h> using namespace std; typedef long long ll; const int N1e55; int t,n,m,a[N],cnt[N]; int main(){scanf("%d",&t);while(t--){scanf("%d%d",&n,&m);for(int i1;i<n;i){scanf(&q…

Mann-Kendall 检验

一、M-K 趋势检验 Mann-Kendall 突变检验是一种非参数的假设检验方法&#xff0c;用于检验时间序列数据中的趋势性变化。该检验方法通过比较每个数据点与其之前数据点的大小&#xff0c;来检测时间序列数据中的单调趋势&#xff08;上升、下降或没有趋势&#xff09;。具体来说…

word转PDF文件变小,图片模糊

word论文29M&#xff0c;文件——另存为——只有1.5M左右&#xff0c;图片压缩严重&#xff0c;图片看不清。 word中很多大图&#xff0c;5M一张的图&#xff0c;所以word很大。 找了很多方法&#xff0c;转换后都在2M左右&#xff0c;勉强可以。 直到找到了这个&#xff0c…

Java-集合类

集合 Java集合是Java中用于存储和管理一组对象的工具。Java集合提供了相应的方法&#xff0c;用于用户对集合内数据的操作。 Java集合类提供了许多不同的数据结构&#xff0c;如列表、队列、栈、集合和映射&#xff0c;以满足不同类型的编程需求。 程序中如何存储大批量同类型…

C 编译原理

C 编译原理 目录 C 编译原理引入GCC 工具链介绍C运行库 编译准备工作编译过程1.预处理2.编译3.汇编4.链接 分析ELF文件1.ELF文件的段2.反汇编ELF C语言编译过程 - 摘录编译预处理编译、优化汇编链接过程 引入 大家肯定都知道计算机程序设计语言通常分为机器语言、汇编语言和高…

(2023 最新版)IntelliJ IDEA 下载安装及配置教程

IntelliJ IDEA下载安装教程&#xff08;图解&#xff09; IntelliJ IDEA 简称 IDEA&#xff0c;由 JetBrains 公司开发&#xff0c;是 Java 编程语言开发的集成环境&#xff0c;具有美观&#xff0c;高效等众多特点。在智能代码助手、代码自动提示、重构、J2EE 支持、各类版本…

深度学习面试八股文(2023.9.06)

一、优化器 1、SGD是什么&#xff1f; 批梯度下降&#xff08;Batch gradient descent&#xff09;&#xff1a;遍历全部数据集算一次损失函数&#xff0c;计算量开销大&#xff0c;计算速度慢&#xff0c;不支持在线学习。随机梯度下降&#xff08;Stochastic gradient desc…

计算机竞赛 机器视觉opencv答题卡识别系统

0 前言 &#x1f525; 优质竞赛项目系列&#xff0c;今天要分享的是 &#x1f6a9; 答题卡识别系统 - opencv python 图像识别 该项目较为新颖&#xff0c;适合作为竞赛课题方向&#xff0c;学长非常推荐&#xff01; &#x1f947;学长这里给一个题目综合评分(每项满分5分…