目录
[一、 交换机持久化 (Exchange Persistence)](#一、 交换机持久化 (Exchange Persistence))
[二、 队列持久化 (Queue Persistence)](#二、 队列持久化 (Queue Persistence))
[三、 消息持久化 (Message Persistence)](#三、 消息持久化 (Message Persistence))
[四、 持久化的"黄金三角" 🔱：三者缺一不可！](#四、 持久化的“黄金三角” 🔱：三者缺一不可！)
[五、 来，完整的代码示例（整合持久化和确认机制）](#五、 来，完整的代码示例（整合持久化和确认机制）)
🌟我的其他文章也讲解的比较有趣😁，如果喜欢博主的讲解方式，可以多多支持一下，感谢🤗！
🌟了解 MQ 请看 ： 【MQ篇】初识MQ！
其他优质专栏： 【🎇SpringBoot】【🎉多线程】【🎨Redis】【✨设计模式专栏（已完结）】...等
如果喜欢作者的讲解方式，可以点赞收藏加关注，你的支持就是我的动力
✨更多文章请看个人主页： 码熔burning
咱们接着聊 RabbitMQ 的"靠谱"修炼之路！上一篇文章讲了生产者确认机制，现在聊聊它的"好记性"------持久化！💾✍️
你想想，咱们电脑里的数据，要么放在内存里（速度快，但关机就没），要么存在硬盘上（速度慢点，但断电还在）。RabbitMQ 也是一样！它处理消息得快，所以很多东西默认是放在内存里的。但如果 RabbitMQ 服务器突然"打个盹"😴 或者"哎呀，摔了一跤"💥（宕机或重启），内存里的东西就"唰"地一下都没了！😱 这可不行，重要的消息和设置怎么能说没就没呢？！
这时候，"持久化"就登场了！它的作用就是把 RabbitMQ 里那些重要的"记忆"写到硬盘上，这样即使服务器重启了，它也能从硬盘上把这些"记忆"读取回来，恢复到宕机前的状态。🐂
RabbitMQ 的持久化，主要涉及三个"记忆点"：
交换机持久化 (Exchange Persistence) - "分拣大厅的名字得刻石头上！" 🏛️
队列持久化 (Queue Persistence) - "每个收件人的信箱要钉牢！" 📬
消息持久化 (Message Persistence) - "信箱里的信要用不褪色的墨水写！" ✍️✉️
来，咱们一个个看，如何在代码里设置它们，以及为啥要这么做！
了解RabbitMQ生产者确认机制请看：【MQ篇】RabbitMQ的生产者消息确认实战！
一、 交换机持久化 (Exchange Persistence)
啥是交换机？ 它就像邮局的分拣大厅入口，负责接收信件（消息）并根据地址（路由键）扔到不同的传送带（队列）上。
为啥要持久化？ 如果交换机不持久化，RabbitMQ 重启后，这个交换机就"失忆"了，不存在了！😳 生产者再往这个名字的交换机发消息就会失败，因为"入口牌子"都没了。
怎么设置？ 在声明（创建）交换机的时候，把 durable 属性设为 true 就行了。
代码怎么写？ 在你的 RabbitConfig 里定义 Exchange 的 Bean 时：
java
复制代码
// 交换机持久化设置
@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 时：
java
复制代码
// 队列持久化设置
@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 方法在内部可能会帮你处理这个细节。但最明确的方式是手动设置消息属性：
java
复制代码
// 消息持久化设置 (在发送消息时)
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() {
@Override
public 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
xml
复制代码
2. application.properties
yml
复制代码
# 配置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
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 {
// 定义一个 ⭐持久化⭐ 主题交换机
@Bean
public TopicExchange myDurableExchange() {
// 参数2: durable = true ✅
System.out.println("🛠️ 正在创建持久化交换机: my.durable.exchange");
return new TopicExchange("my.durable.exchange", true, false);
}
// 定义一个 ⭐持久化⭐ 队列
@Bean
public Queue myDurableQueue() {
// 参数2: durable = true ✅
System.out.println("🛠️ 正在创建持久化队列: my.durable.queue");
return new Queue("my.durable.queue", true, false, false);
}
// 定义一个绑定，将持久化队列绑定到持久化交换机
@Bean
public Binding binding(Queue myDurableQueue, TopicExchange myDurableExchange) {
// 使用固定的路由键 "my.routing.key"
return BindingBuilder.bind(myDurableQueue).to(myDurableExchange).with("my.routing.key");
}
// 配置 RabbitTemplate 并设置 ConfirmCallback
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setMandatory(true);
// ⭐ 设置发送者确认回调 (用于确认消息是否到达 Exchange 并被接受路由) ⭐
rabbitTemplate.setConfirmCallback(new ConfirmCallback() {
@Override
public 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 ⭐
@Bean
public 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 (发送持久化消息)
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 {
@Autowired
private 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() {
@Override
public 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() {
@Override
public 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 (主应用类)
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 {
@Autowired
private 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("👋 应用演示结束.");
}
@Override
public 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 重启后，它会自动尝试重连。如果连接成功，并且消费者容器也配置正确且连接成功，它可能会把队列里的消息消费掉（取决于你的消费者配置）。
通过这个实验，你就能亲眼看到"持久化"的神奇力量：它让你的重要消息熬过了服务器的重启！💪
小提示： 在实际生产环境中，为了确保消息不丢失，发送方确认 + 消费者确认 + 消息持久化 往往是同时使用的，它们从不同环节保障消息的安全。