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

RabbitMQ学习（第二天）

news 来源：原创 2025/8/12 18:02:38

文章目录

1、生产者可靠性
- ①、生产者重连
- ②、生产者确认
2、MQ可靠性
- ①、数据持久化
- ②、LazyQueue(惰性队列)
3、消费者可靠性
- ①、消费者确认
- ②、失败重试机制
- ③、保证业务幂等性
总结

之前的学习中，熟悉了java中搭建和操作RabbitMQ发送接收消息，熟悉使用之后，重点要关注一下面试中常考的点，以及工作经常遇到的问题。今天的学习主要从保证消息可靠性出发，保证可靠性分三部分，分别是生产者可靠性、MQ可靠性、消费者可靠性三部分出发。

1、生产者可靠性

①、生产者重连

有时候因为网络波动，可能导致客户端连接MQ失败，通过配置可以开启连接失败后的重试机制。
yml配置文件：

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

我们故意让它连接不上，结果如下：
在这里插入图片描述

当网络不稳定的时刻，利用重试机制可以有数据显示消息发送的成功率。不过SpringAMQP重试的重复机制是
阻塞式的重试，也就是说多次重试等待的过程是中，当前线程是被阻塞的，会影响到业务性能。

如果对于业务性能有要求，建议参考用重试机制。如果一定需要使用，请合理配置重试等待时长和重试次数，当然也
可以考虑异步用异步多线程来执行发送消息的代码。

②、生产者确认

RabbitMQ有Publisher Confirm和Publisher Return两种确认机制。开启确认机制后，在MQ成功收到消息后会返回确认消息给生产者。返回的结果有以下几种情况：

消息投递到了MQ，但是路由失败。此时会通过PublisherReturn返回路由异常原因，然后返回ACK，告知投递成功
临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
持久消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
其它情况都会返回NACK，告知投递失败

spring:rabbitmq:publisher-confirm-type: correlated # 开启publisher confirm机制，异步confirm类型publisher-returns: true # 开启publisher return机制

publisher-confim:

这里设置 correlated 表示MQ异步回调方式返回回执消息

**publisher-return: **
开启后，会返回路由失败消息。

@Slf4j
@Configuration
public class CommonConfig {@ResourceRabbitTemplate rabbitTemplate;@PostConstructpublic void setRabbitTemplate() {rabbitTemplate.setReturnsCallback(returnedMessage -> {log.error("收到消息的return callback: msg:{} exchange:{}, test:{}, key:{}, code:{}",returnedMessage.getMessage(), returnedMessage.getExchange(), returnedMessage.getReplyText(),returnedMessage.getRoutingKey(), returnedMessage.getReplyCode());});}
}

我们定义一个config类，里面添加一个处理publish-return消息的处理，

测试代码：

    @Testpublic void testConfirmCallBack() throws InterruptedException {CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {@Overridepublic void onFailure(Throwable ex) {log.debug("消息回调失败", ex);}@Overridepublic void onSuccess(CorrelationData.Confirm result) {log.debug("收到confim callback回执");if(result.isAck()) {log.debug("消息发送成功，收到ack");} else{log.debug("消息发送失败，收到nack, 原因：{}", result.getReason());}}});rabbitTemplate.convertAndSend("amq.direct", "red2", "hello", cd);Thread.sleep(2000);}

这里我们测试三种情况：
现有正确的交换机为 amq.direct，routingKey为red，cd逻辑如代码所示。

①、交换机与routingKey均正确，运行结果：

收到confim callback回执
消息发送成功，收到ack

②、交换机正确，routingKey不正确，运行结果：

收到confim callback回执
消息发送成功，收到ack
收到消息的return callback: msg:(Body:'"hello"' MessageProperties [headers={spring_returned_message_correlation=0a8a70fd-a66b-43a3-a681-324e46db7b79, __TypeId__=java.lang.String}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0]) exchange:amq.direct, test:NO_ROUTE, key:red2, code:312

③、交换机不正确，运行结果：

收到confim callback回执
消息发送失败，收到nack, 原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'amq.direct.123' in vhost '/', class-id=60, method-id=40)

关于生产者确认消息要会的几点：

生产者确认需要额外的网络和系统资源开销，尽量不要使用
如果一定要使用，无需开启Publisher-Return机制，因为一般路由失败是自己业务问题
对于nack消息可以有限次数重试，依然失败则记录异常消息

2、MQ可靠性

MO通常将消息保存在内存中，这样可以降低收发消息的延迟，但是会有一些问题：

最常见的问题就是可能因为各种原因宕机重启，而这极大可能会导致消息丢失。
消息堆积在内存中，如果消费者出故障，导致消息积压，引发MQ阻塞。

①、数据持久化

持久化分三部分：

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

前两个的话，Spring中创建时默认将队列和交换机都创建为持久化，当然，手动设置也可以，设置为Durable：
在这里插入图片描述

消息持久化也可以设置：

在这里插入图片描述

我们向他发送了1百万条非持久化消息：
在这里插入图片描述
可以看到其中凹下去的地方就是消息积压导致的。

发送了1百万条持久化消息：
在这里插入图片描述
可以看到此时不存在消息积压。

每次下降的时候，是在将数据写入到磁盘中，但是不会出现宕机为0的情况。

②、LazyQueue(惰性队列)

惰性队列的特征如下：

接收到消息后直接存入磁盘而非内存（内存中只保留最近的消息）
消费者要消费消息时才会从磁盘中读取并加载到内存
支持数百万条的消息存储
在3.12版本后，所有队列都是Lazy Queue模式，无法更改。

控制台创建：
在这里插入图片描述
spring中创建：

@Bean
public Queue lazyQueue() {return QueueBuilder.durable("lazy.queue").lazy() // 开启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);
}

向lazy.queue发送1百万条非持久消息：
在这里插入图片描述
可以看到，性能特别好，因为数据都是直接通过page out到磁盘。

3、消费者可靠性

①、消费者确认

为了确认消费者是否成功处理消息，RabbitMQ提供了消费者确认机制（Consumer Acknowledgement）。当消费者处理消息结束后，应该向RabbitMQ发送一个回执，告知RabbitMQ自己消息处理状态。回执有三种可选值：

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

我们在Spring中配置，auto代表Spring帮我们自动处理这三种情况：

spring:rabbitmq:simple:acknowledge-mode: auto

一般都是直接配置auto。

但是这样就会导致一个问题，如果程序中有异常，那么mq一直重试，这样的话，会导致资源的消耗，对系统造成大的压力，因此我们后面引入了失败重试机制。

②、失败重试机制

配置以下yml内容可以开启失败重试机制：

spring:rabbitmq:listener:simple:prefetch: 1retry:enabled: true #开启失败重试机制initial-interval: 1000ms #初始的失败重试等待时长为1秒multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multipliermax-attempts: 3stateless: true # true无状态，false有状态，如果业务中包含事务，这里为false

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

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

收发消息定义：

@RabbitListener(queues = {"object.queue"})public void ObjectConsumer(String msg) {System.out.println(msg);throw new RuntimeException("......");}@Testpublic void testMessage() {Message message = MessageBuilder.withBody("hello".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT).build();rabbitTemplate.convertAndSend("object.queue", message);}

java代码中实现失败消息处理：


@Slf4j
@Configuration
public class CommonConfig {@ResourceRabbitTemplate rabbitTemplate;@PostConstructpublic void setRabbitTemplate() {rabbitTemplate.setReturnsCallback(returnedMessage -> {log.error("收到消息的return callback: msg:{} exchange:{}, test:{}, key:{}, code:{}",returnedMessage.getMessage(), returnedMessage.getExchange(), returnedMessage.getReplyText(),returnedMessage.getRoutingKey(), returnedMessage.getReplyCode());});}
}

运行结果：
在这里插入图片描述
可以看到报错信息被发送到了error交换机。

③、保证业务幂等性

幂等是一个数学概念，用函数表达式来描述是：f(x) = f(f(x))。在程序开发中，则是指同一个业务，执行一次或多次的结果是一致的。

像查询一般都是幂等的，因为它的执行对于业务状态没有影响，差多少次都没事，就是幂等的。
像用户下单，退单涉及扣钱，退款操作，如果执行多次，会造成经济损失，就是非幂等的。

为了保证幂等性，我们通过一些方法来实现：

(1)、 唯一消息id

是给每个消息都设置一个唯一id，利用id区分是否重复消费：

每一条消息都生成一个唯一的id，与消息一起投递给消费者。
消费者接收到消息后根据自己的业务，业务处理成功后将消息ID保存到数据库库
如果下次又收到相同消息，去数据库查找判断是否存档，存档则为重复消费放弃处理。

@Bean
public MessageConverter messageConverter() {// 1. 定义消息转换器Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();// 2. 配置自动创建消息id，用于识别不同消息，也可以自己在业务中生成ID判断是否重复消费jjmc.setCreateMessageIds(true);return jjmc;
}

但是，这种方案对数据库压力太大，同时速度也相对慢很多，因此只做了解，只需要知道这种方式是jjmc内部帮你创建了一个UUID的随机值作id。

其它方式的话，需要我们结合实际项目在业务逻辑中进行操作来保证幂等性。

总结

最后，用两个问题来总结：

如何保证对服务与交易服务之间的订单状态一致性？

首先，支付服务会在用户支付成功后利用MQ消息通知交易服务，完成订单状态同步。
其次，为了保证MQ消息的可靠性，我们采用了生产者确认机制、消息重确认、消费者失败重试等策略，确保消息投递和处理可靠性。同时也开启了MQ的持久化，避免因服务宕机导致消息丢失。
最后，我们还在交易服务端新增了业务幂等判断，避免重复消费。

如果交易服务消息处理失败，有没有什么兜底方案？

我们可以在交易服务设置定时任务，定期查询订单支付状态。这样即使用MQ通知失败，还可以用定时任务作为兜底方案，确保订单状态最终一致性。

即Spring Task中的@Schedule注解，定时扫描订单状态，用来弥补消息失败，兜底方案。

RabbitMQ学习（第二天）

文章目录 1、生产者可靠性①、生产者重连②、生产者确认 2、MQ可靠性①、数据持久化②、LazyQueue(惰性队列) 3、消费者可靠性①、消费者确认②、失败重试机制③、保证业务幂等性总结之前的学习中，熟悉了java中搭建和操作RabbitMQ发送接收消息，熟悉使用…...

编程日记 2025/8/12 18:02:38

旧版 Flutter 写的项目, 想要在新的环境上运行?

DeepSeek 给出的最佳实践以下是针对拷贝 Flutter 项目到新环境运行的完整检查清单和最佳实践，覆盖了环境配置、版本兼容性、依赖管理等多个关键点： 📋 完整检查清单检查项操作方式/命令重要性1. Flutter SDK 版本flutter --version 对比…...

编程日记 2025/8/12 10:51:03

Flutter接入ProtoBuff和原生Android通信【性能最优】

Protocol Buffers（简称Protobuf）是由 Google 开发的一种结构化数据序列化框架，旨在实现高效的数据交换与存储。其核心特性及优势如下： 一、核心特性 ‌跨语言与跨平台‌ 支持多种编程语言（如 C、Java、Python、Dart …...

编程日记 2025/8/7 20:33:52

【MySQL】（10）用户和权限管理

一、应用场景通常一个应用对应一个数据库，我们希望某个数据库只能被相关人员操纵，就需要创建用户并指定权限。只有登录该用户，才能在权限范围内操纵数据库。root 是权限最高的用户，它拥有所有的权限。二、查询用户在 mysql 数…...

编程日记 2025/8/12 8:46:14

学成在线之缓存

一：缓存把白名单可以看到的信息和学生用户下的我的学习，我的选课等这些信息，存到缓存中，因为这些查询量比较大。当查询时，先去检查缓存中是否有这个数据，如果有，就直接返回如果没有&#…...

编程日记 2025/8/12 17:58:41

视频编解码学习8之视频历史

视频技术的发展历史可以追溯到19世纪，至今已跨越近200年。以下是视频技术发展的主要阶段和里程碑： 1. 早期探索阶段（19世纪-1920年代） 1832年：约瑟夫普拉托（Joseph Plateau）发明"费纳奇镜&…...

编程日记 2025/8/11 16:51:22

从装饰器出发，优雅处理 UI 自动化中的异常

二、装饰器核心原理：函数式编程的“语法糖” 在 UI 自动化测试的实际场景中，页面加载超时、元素定位失败、网络波动等异常问题频繁出现。传统的try-except嵌套方式虽然能捕获异常，但会导致业务代码与处理逻辑高度耦合，造成代码冗…...

编程日记 2025/8/11 20:39:24

华为首款鸿蒙电脑正式亮相

在2025年5月8日的鸿蒙电脑技术与生态沟通会上，华为终端BG平板与PC产品线总裁朱懂东发表演讲。鸿蒙办公将支持无缝协同（统一账户 / 多端同步）、智慧体验（鸿蒙 AI / 智慧交互）、融合生态（桌面生态 / 移动生态…...

编程日记 2025/8/12 17:54:44

森林生态学研究深度解析：R语言入门、生物多样性分析、机器学习建模与群落稳定性评估

在生态学研究中，森林生态系统的结构、功能与稳定性是核心研究内容之一。这些方面不仅关系到森林动态变化和物种多样性，还直接影响森林提供的生态服务功能及其应对环境变化的能力。森林生态系统的结构主要包括物种组成、树种多样性、树木的空间分布与密度…...

编程日记 2025/8/7 22:06:39

需求分析阶段测试工程师主要做哪些事情

在软件测试需求分析阶段，主要围绕确定测试范围、明确测试目标、细化测试内容等方面开展工作，为后续测试计划的制定、测试用例的设计以及测试执行提供清晰、准确的依据。以下是该阶段具体要做的事情： 1. 需求收集与整理收集需求文档&#x…...

编程日记 2025/8/12 17:54:43

IBM BAW（原BPM升级版）使用教程第五讲

结前篇！ 一、服务：外部服务在 IBM Business Automation Workflow (BAW) 中，外部服务（External Services）是指在流程中调用和集成外部系统或服务的组件。外部服务允许IBM BAW与其他业务系统、应用程序或第三方服务进行…...

编程日记 2025/8/12 8:26:02

一个用C#开发的记事本Notepads开源编辑器

从零学习构建一个完整的系统推荐一个用C#开发的，模仿Nodepad的记事本编辑器。项目简介这是一个国人开发的，采用Fluent Design风格，支持Markdown语法及其预览，可以用于替代Windows自带记事本功能。虽然它没有Nodepad、VS C…...

编程日记 2025/8/12 17:56:43

Office宏病毒钓鱼-打点突破

Office宏钓鱼 [攻击路径] / [利用流程] - 详细步骤：制作一个简单的Word宏钓鱼文档步骤一：规划恶意宏功能 (Payload Design) 目的： 决定宏执行后要达成的目标。为教学演示，我们设定一个简单的目标：弹出一个消息框&…...

编程日记 2025/8/12 17:53:47

神经网络—感知器、多层感知器

文章目录前言一、生物神经元与感知器的类比二、感知器1、简单感知器2、多层感知器（1）多层感知机结构 3、神经网络结构总结1、感知器的局限性如何突破感知器的局限性？ 2、感知器的应用前言感知器（Perceptron）是神经…...

编程日记 2025/8/9 22:25:21

美化IDEA注释：Idea 中快捷键 Ctrl + / 自动注释的缩进（避免添加注释自动到行首）以及 Ctrl + Alt + l 全局格式化代码的注释缩进

打开 Settings 界面，依次选择 Editor -> Code Style -> Java，选择 Code Generation， 取消 Line comment at first column 和 Block comment at first column 的勾选即可， 1、Line comment at first column (行注释在第一列…...

编程日记 2025/8/12 17:54:43

基于大型语言模型的高效时间序列预测模型选择

时间序列预测在金融、医疗、能源等多个领域具有重要意义，其结果对于决策和规划起到关键作用。然而，选择合适的预测模型往往需要丰富的领域知识和大量计算资源。Abdallah等（2022）的研究表明，没有单一算法能在所有预测任…...

编程日记 2025/8/12 17:53:44

2025 EAU UTUC指南学习笔记③：诊断策略精读——从症状到活检，如何科学判断治疗路径？

👋 欢迎关注我的学习专栏~ 如果觉得文章有帮助，别忘了点赞、关注、评论，一起学习本文聚焦UTUC的诊断策略，全面梳理影像学检查、尿液学检测、内镜评估与活检手段，并结合指南推荐等级，提出科研视角的思考。 …...

编程日记 2025/8/11 22:37:24

【Ubuntu】Netplan静态网络配置

1.配置文件 # 默认配置文件：/etc/netplan/*.yaml# 本机 rootubuntu-server:~# vim /etc/netplan/50-cloud-init.yaml 2.DHCP network: ethernets:ens33:dhcp4: trueversion: 2 3.静态IP设置注意：netplan 说明文件格式存储在下列路径下…...

编程日记 2025/8/9 16:23:22

【SQLSERVER】Ubuntu 连接远程 SQL Server（MSSQL）

Ubuntu 连接远程 SQL Server（MSSQL） 引言在企业级开发环境中，跨平台的数据库连接是常见需求。虽然 SQL Server 通常在 Windows 服务器上运行，但在 Ubuntu 上连接远程 SQL Server 也非常重要。本文将详细介绍如何在 Ubuntu 上通…...

编程日记 2025/8/9 15:26:36

GuPPy-v1.2.0安装与使用-生信工具52

GuPPy：Python中用于光纤光度数据分析的免费开源工具 01 背景 Basecalling 是将原始测序信号转换为碱基序列的过程，通俗地说，就是“把碱基识别出来”。这一过程在不同代测序技术中各不相同： 一代测序是通过解析峰图实现&#xff1…...

编程日记 2025/8/8 0:13:32

Mysql-OCP PPT课程讲解并翻译

#跳过介绍，直接从干货开始记录第一章安装mysql...

编程日记 2025/8/10 13:03:28

异常如下 Description:An attempt was made to call a method that does not exist. The attempt was made from the following location:com.google.common.cache.CacheBuilder.initialCapacity(CacheBuilder.java:353)The following method did not exist:com.google.common…...

编程日记 2025/8/10 17:30:55