【MQ篇】RabbitMQ之消息持久化！

在这里插入图片描述

你想想，咱们电脑里的数据，要么放在内存里（速度快，但关机就没），要么存在硬盘上（速度慢点，但断电还在）。RabbitMQ 也是一样！它处理消息得快，所以很多东西默认是放在内存里的。但如果 RabbitMQ 服务器突然“打个盹”😴 或者“哎呀，摔了一跤”💥（宕机或重启），内存里的东西就“唰”地一下都没了！😱 这可不行，重要的消息和设置怎么能说没就没呢？！

这时候，“持久化”就登场了！它的作用就是把 RabbitMQ 里那些重要的“记忆”写到硬盘上，这样即使服务器重启了，它也能从硬盘上把这些“记忆”读取回来，恢复到宕机前的状态。🐂

RabbitMQ 的持久化，主要涉及三个“记忆点”：

交换机持久化 (Exchange Persistence) - “分拣大厅的名字得刻石头上！” 🏛️
队列持久化 (Queue Persistence) - “每个收件人的信箱要钉牢！” 📬
消息持久化 (Message Persistence) - “信箱里的信要用不褪色的墨水写！” ✍️✉️

来，咱们一个个看，如何在代码里设置它们，以及为啥要这么做！

了解RabbitMQ生产者确认机制请看：【MQ篇】RabbitMQ的生产者消息确认实战！

一、交换机持久化 (Exchange Persistence)

啥是交换机？ 它就像邮局的分拣大厅入口，负责接收信件（消息）并根据地址（路由键）扔到不同的传送带（队列）上。
为啥要持久化？ 如果交换机不持久化，RabbitMQ 重启后，这个交换机就“失忆”了，不存在了！😳 生产者再往这个名字的交换机发消息就会失败，因为“入口牌子”都没了。
怎么设置？ 在声明（创建）交换机的时候，把 durable 属性设为 true 就行了。
代码怎么写？ 在你的 RabbitConfig 里定义 Exchange 的 Bean 时：

// 交换机持久化设置
@Bean
public TopicExchange myDurableExchange() {// 参数1: 交换机名称// 参数2: durable 是否持久化，设为 true！✅// 参数3: autoDelete 是否自动删除，设为 false (通常不自动删除持久化交换机)System.out.println("🛠️ 正在创建持久化交换机: my.durable.exchange");return new TopicExchange("my.durable.exchange", true, false);
}

通过设置第二个参数为 true，你就告诉 RabbitMQ：“喂，这个叫 my.durable.exchange 的交换机很重要，给我记到小本本上！” ✍️

在这里插入图片描述

注意：默认情况下，由SpringAMQP声明的交换机都是持久化的。

二、队列持久化 (Queue Persistence)

啥是队列？ 它就像收件人专属的信箱，消息最终会待在这里，等着被消费者取走。
为啥要持久化？ 如果队列不持久化，RabbitMQ 重启后，这个队列也“失忆”了，不见了！😵‍💫 不仅队列本身没了，更要命的是，如果这个队列里当时还存着没被消费的消息，它们也会跟着一起消失！ 🗑️ 这绝对是消息丢失的重大风险点！
怎么设置？ 在声明（创建）队列的时候，把 durable 属性设为 true。
代码怎么写？ 在你的 RabbitConfig 里定义 Queue 的 Bean 时：

// 队列持久化设置
@Bean
public Queue myDurableQueue() {// 参数1: 队列名称// 参数2: durable 是否持久化，设为 true！✅// 参数3: exclusive 是否独占 (通常不独占)// 参数4: autoDelete 是否自动删除System.out.println("🛠️ 正在创建持久化队列: my.durable.queue");return new Queue("my.durable.queue", true, false, false);
}

把第二个参数设为 true，你就给这个队列上了把“锁” 🔒，告诉 RabbitMQ：“这个信箱要钉死在这儿，重启了也得给我留着！”

在这里插入图片描述

注意：默认情况下，由SpringAMQP声明的队列都是持久化的。

三、消息持久化 (Message Persistence)

啥是消息？ 就是你在队列里放着的那些“信件”本身的内容。
为啥要持久化？ 前面说了，队列持久化能保证队列这个“信箱”不消失。但是！ 如果信箱里的“信件”本身没有做持久化处理（默认是放在内存里），那么即使信箱（队列）还在，里面的信件也会在 RabbitMQ 重启时“蒸发”！🌫️ 所以，为了让消息在持久化队列中真正“活下来”，消息本身也必须是持久化的！
怎么设置？ 在发送消息的时候，设置消息的 deliveryMode 属性为 persistent (模式 2)。
代码怎么写？ 在 Spring AMQP 中，你可以通过 MessagePostProcessor 来修改消息属性，或者通常情况下，如果你往一个持久化的队列发送消息，convertAndSend 方法在内部可能会帮你处理这个细节。但最明确的方式是手动设置消息属性：

// 消息持久化设置 (在发送消息时)
public void sendPersistentMessage(String message) {CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());System.out.println("📨 正在发送持久化消息: '" + message + "', ID: " + correlationData.getId());rabbitTemplate.convertAndSend("my.durable.exchange", "my.routing.key", message, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {// ⭐ 设置消息的投递模式为持久化 (DeliveryMode.PERSISTENT) ⭐message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);System.out.println("✉️ 消息已标记为持久化!");return message;}}, correlationData); //别忘了带上 correlationData 给 ConfirmCallback 用！System.out.println("📬 持久化消息已提交到 RabbitTemplate，等待 RabbitMQ ACK...");
}

MessagePostProcessor 允许你在消息发送前“拦截”并修改消息的属性。这里我们就把它设置为 DeliveryMode.PERSISTENT。这就像在你写好的信封上盖个大红章 💌：“此件重要！请务必保存！”

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

四、持久化的“黄金三角” 🔱：三者缺一不可！

理解持久化，最关键的是记住这个“黄金三角”：

只有当交换机是持久化的 (durable exchange) + 队列是持久化的 (durable queue) + 消息是持久化的 (persistent message)，消息才能在 RabbitMQ 重启后依然保存在队列中！

如果队列不持久化，交换机和消息再怎么持久化也没用，队列没了，一切皆空。☁️
如果队列持久化了，但消息不持久化，重启后队列还在，但里面的消息都没了。💧
如果消息持久化了，但队列不持久化，同样队列没了消息也无法保存。
交换机持久化主要是为了确保 Exchange 本身重启后还在，否则生产者都找不到它发消息了。

所以，为了确保消息在 RabbitMQ 宕机重启时不丢失，这三兄弟（持久化交换机、持久化队列、持久化消息）得协同工作才行！🤝

五、来，完整的代码示例（整合持久化和确认机制）

咱们在之前的代码基础上，把持久化设置加进去，并且依然保留发送者确认机制，这俩功能是保护消息的左膀右臂！💪

1. pom.xml

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2. application.properties

# 配置Spring应用中的RabbitMQ连接参数
spring:rabbitmq:# RabbitMQ服务器的主机地址host: localhost# RabbitMQ服务器的端口号port: 5672# 访问RabbitMQ服务器的用户名username: guest# 访问RabbitMQ服务器的密码password: guest# 配置发布确认的类型为correlated，以便在消息发送后收到确认publisher-confirm-type: correlated# 启动返回机制，当消息无法投递时返回给发送者publisher-returns: true# 配置RabbitMQ模板的参数template:# 设置所有消息都是必须投递的mandatory: true# 设置等待回复的超时时间为60000毫秒reply-timeout: 60000# 配置日志级别
logging:level:# 设置org.springframework.amqp包下的日志级别为DEBUG，以便捕获AMQP相关的调试信息org:springframework:amqp: DEBUG

3. RabbitConfig.java

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.lang.Nullable;@Configuration
public class RabbitConfig {// 定义一个 ⭐持久化⭐ 主题交换机@Beanpublic TopicExchange myDurableExchange() {// 参数2: durable = true ✅System.out.println("🛠️ 正在创建持久化交换机: my.durable.exchange");return new TopicExchange("my.durable.exchange", true, false);}// 定义一个 ⭐持久化⭐ 队列@Beanpublic Queue myDurableQueue() {// 参数2: durable = true ✅System.out.println("🛠️ 正在创建持久化队列: my.durable.queue");return new Queue("my.durable.queue", true, false, false);}// 定义一个绑定，将持久化队列绑定到持久化交换机@Beanpublic Binding binding(Queue myDurableQueue, TopicExchange myDurableExchange) {// 使用固定的路由键 "my.routing.key"return BindingBuilder.bind(myDurableQueue).to(myDurableExchange).with("my.routing.key");}// 配置 RabbitTemplate 并设置 ConfirmCallback@Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);rabbitTemplate.setMandatory(true);// ⭐ 设置发送者确认回调 (用于确认消息是否到达 Exchange 并被接受路由) ⭐rabbitTemplate.setConfirmCallback(new ConfirmCallback() {@Overridepublic void confirm(@Nullable CorrelationData correlationData, boolean ack, @Nullable String cause) {String messageId = correlationData != null ? correlationData.getId() : "N/A";if (ack) {System.out.println("✨ RabbitMQ 确认收到消息并处理成功！Message ID: " + messageId + " (Ack: " + ack + ")");} else {System.err.println("💔 RabbitMQ 拒绝或未能处理消息！Message ID: " + messageId + ", 原因: " + cause + " (Ack: " + ack + ")");}}});// 这里省略了 ReturnCallback，它用于处理路由不到队列的消息 (需要 mandatory=true)// rabbitTemplate.setReturnsCallback(...)return rabbitTemplate;}// 简单的消费者（非持久化核心，用于演示消息能到达）// ⭐ 注意：这个消费者不会做消费者确认，消息消费后可能还在队列里，直到被 RabbitMQ 删除 ⭐// ⭐ 真正的持久化测试需要消费者确认并手动 ack ⭐@Beanpublic org.springframework.amqp.rabbit.listener.MessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory, Queue myDurableQueue) {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueues(myDurableQueue);container.setMessageListener(message -> {System.out.println("👂 消费者收到消息: '" + new String(message.getBody()) + "'");// 在实际应用中，这里处理完消息后需要手动发送 ACK 给 RabbitMQ// 如果不发 ACK 且设置了手动确认模式，消息会一直留在队列直到连接断开并重发// 这里为了简单演示持久化队列里有消息，设置为 NO_ACK 模式});// ⭐ 注意：AcknowledgeMode.NONE (不确认) 模式下，RabbitMQ 会立即删除消息，不利于演示重启后消息还在 ⭐// ⭐ 为了演示消息在队列中持久化，建议消费者代码暂时注释掉或者设置为手动确认且不手动 ack ⭐// ⭐ 本例代码中暂时保留消费者，但请注意其行为，真正的持久化测试需要手动停止消费者、重启RabbitMQ、再启动消费者查看 ⭐container.setAcknowledgeMode(AcknowledgeMode.NONE); // 简单演示，不处理消费者确认return container;}
}

4. MessageSender.java (发送持久化消息)

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.util.UUID;@Component
public class MessageSender {@Autowiredprivate RabbitTemplate rabbitTemplate;// 使用持久化的交换机和队列名称private static final String EXCHANGE_NAME = "my.durable.exchange";private static final String QUEUE_NAME = "my.durable.queue"; // 虽然发送不需要队列名，这里记一下private static final String ROUTING_KEY = "my.routing.key";private static final String NON_EXISTENT_EXCHANGE = "non.existent.exchange";/*** 发送一条 ⭐持久化⭐ 消息到 ⭐持久化⭐ 交换机和队列* @param message 需要发送的消息内容*/public void sendPersistentMessage(String message) {CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());System.out.println("📨 正在发送持久化消息: '" + message + "', ID: " + correlationData.getId());rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY, message, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {// ⭐ 核心：设置消息的投递模式为持久化 ⭐message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);System.out.println("✉️ 消息属性已设置为持久化!");return message;}}, correlationData);System.out.println("📬 持久化消息已提交到 RabbitTemplate，等待 RabbitMQ ACK...");}/*** 演示发送失败的情况，发送到一个不存在的 Exchange，预期会收到 NACK* 依然使用持久化属性，但因为 Exchange 不存在，消息无法到达 Exchange* @param message 需要发送的消息内容*/public void sendFailedPersistentMessage(String message) {CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());System.out.println("😠 尝试发送持久化消息到不存在的 Exchange: '" + NON_EXISTENT_EXCHANGE + "'");System.out.println("📨 正在发送失败消息: '" + message + "', ID: " + correlationData.getId());// 发送消息到不存在的交换机，并尝试标记为持久化（但消息根本到不了Exchange）rabbitTemplate.convertAndSend(NON_EXISTENT_EXCHANGE, ROUTING_KEY, message, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {// 虽然设置了持久化，但这消息因为 Exchange 不存在，根本不会被 RabbitMQ 接收并存盘message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);return message;}}, correlationData);System.out.println("📬 失败消息已提交到 RabbitTemplate，等待 RabbitMQ NACK...");}
}

5. Application.java (主应用类)

import com.gewb.produce_confire.MessageSender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;import java.util.concurrent.TimeUnit;@SpringBootApplication
public class Application implements CommandLineRunner {@Autowiredprivate MessageSender messageSender;public static void main(String[] args) {SpringApplication.run(Application.class, args);// ⭐ 重要：为了让异步回调和持久化演示效果有时间展现，主线程不要马上退出 ⭐// 在非 Web 应用中，需要手动保持应用运行一段时间try {System.out.println("😴 应用正在运行，等待回调和可能的消费者消费。请手动停止 RabbitMQ 后再启动进行持久化测试。");TimeUnit.SECONDS.sleep(20); // 延长等待时间，以便观察日志并有时间手动操作 RabbitMQ} catch (InterruptedException e) {Thread.currentThread().interrupt();System.err.println("😴 应用被中断唤醒.");}System.out.println("👋 应用演示结束.");}@Overridepublic void run(String... args) throws Exception {System.out.println("🚀 应用启动，开始发送测试持久化消息...");// 1. 发送一条会成功投递到持久化队列的持久化消息messageSender.sendPersistentMessage("Important Persistent Message!");// 等待一下，让第一条消息的确认回调先执行TimeUnit.SECONDS.sleep(2);// 2. 发送一条会失败的消息（发送到不存在的 Exchange），看 NACK 回调messageSender.sendFailedPersistentMessage("This persistent message should be NACKED!");System.out.println("✅ 所有测试消息已发送提交。请观察日志输出中的 ACK/NACK 回调结果。");System.out.println(">>> 要测试持久化效果，请在看到 ACK 日志后，手动停止 RabbitMQ 服务器，然后重新启动，检查队列中是否有 'Important Persistent Message!' 这条消息。");}
}

运行结果：

在这里插入图片描述

注意，在 main 方法里，我加了更长的延时 (20 秒)，并提示用户手动停止和启动 RabbitMQ 来测试持久化效果。

如何测试持久化效果？

确保 RabbitMQ 运行中。
运行上面的 Spring Boot 应用。
在控制台看到 ✨ RabbitMQ 确认收到消息并处理成功！Message ID: ... (ACK) 日志后，手动停止 RabbitMQ 服务器。
检查 RabbitMQ 的数据目录（如果你知道在哪儿），可以看到有一些文件生成了（这些就是持久化数据）。
重新启动 RabbitMQ 服务器。
使用 RabbitMQ Management Plugin (如果安装了) 或其他客户端工具，查看 my.durable.queue 这个队列。你会发现之前发送的 Important Persistent Message! 这条消息仍然在队列里！🥳 而发送到不存在 Exchange 的消息，因为根本没被 RabbitMQ 接受，当然不会出现在任何队列里。
如果你的 Spring Boot 应用还没退出（因为设置了 20 秒延时），RabbitMQ 重启后，它会自动尝试重连。如果连接成功，并且消费者容器也配置正确且连接成功，它可能会把队列里的消息消费掉（取决于你的消费者配置）。

通过这个实验，你就能亲眼看到“持久化”的神奇力量：它让你的重要消息熬过了服务器的重启！💪

小提示： 在实际生产环境中，为了确保消息不丢失，发送方确认 + 消费者确认 + 消息持久化 往往是同时使用的，它们从不同环节保障消息的安全。