【MQ篇】RabbitMQ的消费者确认机制实战！
目录
- 一、啥是消费者确认机制？
- 二、消费者可以给 RabbitMQ 发啥信号？
- 三、RabbitMQ 提供哪几种“签字”模式？
- 四、代码怎么设置手动确认？
- 五、完整的代码示例（整合消费者确认）
🌟我的其他文章也讲解的比较有趣😁，如果喜欢博主的讲解方式，可以多多支持一下，感谢🤗！
🌟了解 MQ 请看 ： 【MQ篇】初识MQ！
其他优质专栏： 【🎇SpringBoot】【🎉多线程】【🎨Redis】【✨设计模式专栏（已完结）】…等
如果喜欢作者的讲解方式，可以点赞收藏加关注，你的支持就是我的动力
✨更多文章请看个人主页： 码熔burning
接着咱们的可靠性系列，前面讲了确保消息“出了门”并且“写到了账本上”（发送方确认和持久化）。现在，咱们要聊聊消息可靠性的最后一公里：确保消息被“收件人”成功处理了！💌🏠✅
你想想，快递小哥把包裹送到你家门口，往地上一放，咔嚓拍个照 📸，就算“投递成功”了。但如果家里没人，包裹被风吹走了 💨，或者被隔壁老王的狗叼走了 🐶📦💥，或者你拿到包裹后，手一滑掉粪坑里了 🚽… 总之，包裹是“投递”了，但你压根没用到里面的东西！
在 RabbitMQ 里，“消费者”（Consumer）就是那个“收件人”。RabbitMQ 把消息发给消费者，如果消费者还没来得及处理完就“嗝屁了” 😵💫（程序崩溃、网络断开、处理异常），那这条消息对于 RabbitMQ 来说可能已经“送达”了，然后就被无情地从队列里删除了！🗑️ 等消费者重启后，它再也收不到那条“丢失”的消息了。这绝对是大问题！
了解生产者确认机制请看：【MQ篇】RabbitMQ的生产者消息确认实战！
了解消息持久化请看：【MQ篇】RabbitMQ之消息持久化！
一、啥是消费者确认机制？
消费者确认机制就是为了解决这个问题而生的！它是一套“收件人签字”系统。📦✍️ RabbitMQ 把消息发给消费者后，并不会立即从队列里删除它。它会等着消费者给它一个明确的信号：“喂，这个消息我收到啦，并且处理完了！ ✅” 或者 “这个消息我收到啦，但是处理失败了！ ❌”
只有当 RabbitMQ 收到消费者发来的“处理成功”信号后，它才会放心地把这条消息从队列里永久删除。如果在收到确认信号之前，消费者断开连接了，或者明确表示处理失败了，RabbitMQ 就会知道这条消息“有问题”，会考虑把这条消息重新发给其他消费者，或者等这个消费者恢复后再发给它。
二、消费者可以给 RabbitMQ 发啥信号？
消费者可以给 RabbitMQ 发送以下几种“签字”信号：
basic.ack(确认成功) ✅： “这个消息我彻底处理完了，非常满意！你可以从队列里删了！” 这是最常见的信号。basic.nack(否定确认) ❌ /basic.reject(拒绝) ✋： “这个消息我处理不了！” 这两个信号都可以表示处理失败。basic.reject只能拒绝一条消息。basic.nack可以批量拒绝（虽然不常用）。- 这两个信号都有一个重要的参数：
requeue。requeue = true🔄： “我处理不了，但这不是消息的问题，可能是我的临时问题，或者我觉得其他消费者能处理。麻烦你把这条消息重新放回队列吧！”requeue = false🚮： “我处理不了，而且这条消息本身可能有问题，或者我不想再看到它了。请丢弃它吧！” （或者根据配置发到死信队列 DLX）
三、RabbitMQ 提供哪几种“签字”模式？
在 Spring AMQP 里，你可以设置消费者容器（MessageListenerContainer）的确认模式：
AcknowledgeMode.NONE(自动确认) 🏃💨： 最不安全！ ❌ RabbitMQ 把消息发给消费者后，立即就把它从队列里删了，不等消费者处理结果。这就像快递员把包裹往门口一扔就跑路，根本不管你有没有拿到或处理。如果消费者收到消息后还没处理完就崩了，消息就彻底丢了！除了对消息丢失不敏感的场景，强烈不建议使用！AcknowledgeMode.AUTO(智能自动确认) 🧠： 比NONE安全一点。Spring AMQP 会监听你的消费者方法执行情况。如果消费者方法成功返回且没有抛出异常，Spring AMQP 会帮你发一个basic.ack。如果消费者方法抛出异常，Spring AMQP 会帮你发一个basic.nack(通常requeue=true)。这个模式在很多场景下够用了，但它基于方法是否抛异常，如果你的业务逻辑处理失败但不抛异常，它还是会发 ACK。AcknowledgeMode.MANUAL(手动确认) ✍️🔒： 最安全也最灵活！ 👍 RabbitMQ 把消息发给消费者后，会一直等着。消费者必须自己代码里调用 Channel 的basicAck或basicNack/basicReject方法来明确告知 RabbitMQ 处理结果。只有收到了手动发出的basic.ack，RabbitMQ 才会删除消息。如果消费者崩溃了，或者忘记发送确认信号，RabbitMQ 会认为消息未被确认，并在连接断开后重新投递（默认行为）。这是处理重要消息的推荐模式！
四、代码怎么设置手动确认？
要使用 AcknowledgeMode.MANUAL，你需要：
-
设置消费者容器的确认模式：
// 在 RabbitConfig 或消费者配置类里 container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // ⭐设置为手动确认模式⭐如果使用其他两种模式，可以在配置文件中设置：
spring:rabbitmq:listener:simple:acknowledge-mode: none #或者auto -
修改消费者监听方法的签名： 你的方法需要能够获取到 RabbitMQ 的 Channel 对象和消息的
deliveryTag（投递标签）。deliveryTag是 RabbitMQ 为每个 Channel 上的每次投递分配的唯一标识符。你需要用它来告诉 RabbitMQ 你确认的是哪条消息。// 消费者类里，监听消息的方法 // 注意参数：Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag public void receiveMessage(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {String msgContent = new String(message.getBody());System.out.println("👂 消费者收到消息: '" + msgContent + "', Delivery Tag: " + deliveryTag);try {// ⭐ 在这里处理你的业务逻辑 ⭐boolean processSuccess = processMyBusiness(msgContent); // 假设这是你的业务处理方法if (processSuccess) {// ⭐ 业务处理成功！手动发送 ACK ⭐channel.basicAck(deliveryTag, false); // 参数2: multiple=false，只确认当前这一条System.out.println("✅ 消息处理成功，手动发送 ACK！Delivery Tag: " + deliveryTag);} else {// ⭐ 业务处理失败！手动发送 NACK/Reject ⭐// channel.basicReject(deliveryTag, true); // Reject 只能拒绝单条channel.basicNack(deliveryTag, false, true); // 参数2: multiple=false；参数3: requeue=true 重新入队System.err.println("❌ 消息处理失败，手动发送 NACK！Delivery Tag: " + deliveryTag + ", 将重新入队。");}} catch (Exception e) {// ⭐ 处理过程中抛出异常，通常也发送 NACK，并决定是否重新入队 ⭐System.err.println("🚨 处理消息时发生异常！Delivery Tag: " + deliveryTag + ", 错误: " + e.getMessage());try {// 异常时发送 NACK，并重新入队channel.basicNack(deliveryTag, false, true); // requeue=trueSystem.err.println("💔 发生异常，手动发送 NACK 并重新入队！Delivery Tag: " + deliveryTag);} catch (Exception ex) {System.err.println("🔥 发送 NACK 时也发生异常！ Delivery Tag: " + deliveryTag + ", 错误: " + ex.getMessage());// 这里可能需要记录日志或采取其他措施，这条消息可能丢失}} }// 模拟业务处理方法，有时候成功，有时候失败 private boolean processMyBusiness(String msg) {// 假设消息内容包含 "fail" 关键字就模拟失败if (msg.toLowerCase().contains("fail")) {System.out.println("故意模拟业务处理失败: " + msg);return false;}// 模拟耗时操作try {TimeUnit.MILLISECONDS.sleep(200);} catch (InterruptedException e) {Thread.currentThread().interrupt();}return true; }在这个例子里，我们：
- 方法签名里加了
Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag。 processMyBusiness模拟业务处理，包含一个模拟失败的逻辑 ("fail"关键字)。- 根据
processMyBusiness的返回值，手动调用channel.basicAck或channel.basicNack。 - 在
basicNack中，我们设置了requeue=true，这意味着处理失败的消息会被放回队列，稍后可能被重新投递。 - 还加了异常处理，如果业务处理过程中抛异常，也发送 NACK。
- 方法签名里加了
五、完整的代码示例（整合消费者确认）
咱们基于之前的持久化代码，把消费者部分修改为手动确认模式，并演示 ACK 和 NACK (requeue=true) 的效果。
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.amqp.rabbit.listener.api.ChannelAwareMessageListener; // 引入 ChannelAwareMessageListener 接口import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.lang.Nullable;@Configuration
public class RabbitConfig {// 持久化交换机 (同上)@Beanpublic TopicExchange myDurableExchange() {System.out.println("🛠️ 正在创建持久化交换机: my.durable.exchange");return new TopicExchange("my.durable.exchange", true, false);}// 持久化队列 (同上)@Beanpublic Queue myDurableQueue() {System.out.println("🛠️ 正在创建持久化队列: my.durable.queue");return new Queue("my.durable.queue", true, false, false);}// 绑定 (同上)@Beanpublic Binding binding(Queue myDurableQueue, TopicExchange myDurableExchange) {return BindingBuilder.bind(myDurableQueue).to(myDurableExchange).with("my.routing.key");}// RabbitTemplate 配置 (含 Publisher Confirms) (同上)@Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);rabbitTemplate.setMandatory(true);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("✨ Publisher Confirm: 收到消息 ACK！Message ID: " + messageId);} else {System.err.println("💔 Publisher Confirm: 收到消息 NACK！Message ID: " + messageId + ", 原因: " + cause);}}});return rabbitTemplate;}// ⭐ 配置手动确认的消费者容器，使用 ChannelAwareMessageListener ⭐@Beanpublic SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory, Queue myDurableQueue, ChannelAwareMessageListener manualAckListener) { // 注入我们实现的监听器 BeanSimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueues(myDurableQueue); // 监听持久化队列// ⭐ 核心：设置我们实现的 ChannelAwareMessageListener ⭐container.setMessageListener(manualAckListener);// ⭐ 必须设置确认模式为手动确认！ ⭐container.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 可以设置预取计数，手动确认模式下常用container.setPrefetchCount(5); // 例子：一次最多拉取 5 条未确认消息System.out.println("👂 消费者容器配置完成，模式: 手动确认 (MANUAL)，使用 ChannelAwareMessageListener");return container;}// ⭐ 声明实现了 ChannelAwareMessageListener 接口的 Bean ⭐@Beanpublic ChannelAwareMessageListener manualAckListener() {return new ManualAckMessageListener(); // 返回我们实现类的一个实例}
}
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 ROUTING_KEY = "my.routing.key";private static final String NON_EXISTENT_EXCHANGE = "non.existent.exchange";// 发送持久化消息到持久化 Exchange/Queuepublic 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 (同上)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());rabbitTemplate.convertAndSend(NON_EXISTENT_EXCHANGE, ROUTING_KEY, message, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);return message;}}, correlationData);System.out.println("📬 失败消息已提交到 RabbitTemplate，等待 RabbitMQ NACK...");}
}
5. ManualAckMessageListener.java (新的消费者类，包含手动确认逻辑)
新建一个类 ManualAckMessageListener.java：
import com.rabbitmq.client.Channel; // 引入 RabbitMQ 的 Channel 类
import org.springframework.amqp.core.Message; // 引入 Spring AMQP 的 Message 类
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; // 引入接口
import org.springframework.stereotype.Component; // Component 注解让 Spring 扫描到它 (虽然在 Config 里手动 Bean 了，加了也无妨)import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;// ⭐ 实现 ChannelAwareMessageListener 接口 ⭐
public class ManualAckMessageListener implements ChannelAwareMessageListener {// ⭐ 依然用作总尝试次数的计数器 ⭐private AtomicInteger processAttemptCount = new AtomicInteger(0);/*** ⭐ 实现 onMessage 方法，直接获取 Message 和 Channel ⭐* 这是 ChannelAwareMessageListener 的核心方法，RabbitMQ 收到消息后会调用它* @param message 收到的原始消息对象* @param channel RabbitMQ 的 Channel 对象，用于发送 ACK/NACK 等指令*/@Overridepublic void onMessage(Message message, Channel channel) {long deliveryTag = message.getMessageProperties().getDeliveryTag();String msgContent = new String(message.getBody());// ⭐ 核心变化：先自增，获取本次是第几次尝试处理 ⭐int currentAttempt = processAttemptCount.incrementAndGet();System.out.println("\n---");System.out.println("👂 消费者收到消息: '" + msgContent + "', Delivery Tag: " + deliveryTag + ", Redelivered: " + message.getMessageProperties().isRedelivered() + ", 这是总第 " + currentAttempt + " 次尝试处理消息.");try {// ⭐ 模拟业务处理逻辑：我们让包含 "fail" 的消息在前两次尝试时失败 ⭐// 这里的逻辑是：如果消息包含 "fail" 且这是第 1 次 或 第 2 次尝试 (总的尝试次数)，则模拟失败if (msgContent.contains("fail") && currentAttempt <= 2) {System.out.println("故意模拟业务处理失败: 这是第 " + currentAttempt + " 次尝试");throw new RuntimeException("模拟业务处理失败"); // 抛出异常触发失败逻辑}// ⭐ 业务处理成功！ ⭐// 模拟正常的业务处理时间TimeUnit.MILLISECONDS.sleep(500);System.out.println("✅ 业务处理成功！消息内容: '" + msgContent + "', 这是第 " + currentAttempt + " 次尝试.");// ⭐ 业务处理成功，手动发送 ACK ⭐channel.basicAck(deliveryTag, false);System.out.println("👍 消息处理成功，手动发送 ACK！Delivery Tag: " + deliveryTag);// 注意：这里不再重置 processAttemptCount。它现在统计的是这个监听器 Bean 总共处理了多少次消息投递尝试。// 如果你需要一个针对“某条特定消息”的重试计数，逻辑会复杂得多，需要通过消息的唯一ID（比如 correlationData 或消息体里的业务ID）配合一个 Map 来追踪。// 但对于演示 NACK -> Requeue -> 再次被同一个（或另一个）消费者实例收到并最终处理成功，这个全局计数器就足够了。} catch (Exception e) {// ⭐ 业务处理失败或发生异常 ⭐System.err.println("❌ 业务处理失败或发生异常！消息内容: '" + msgContent + "', Delivery Tag: " + deliveryTag + ", 这是第 " + currentAttempt + " 次尝试, 错误: " + e.getMessage());try {// ⭐ 手动发送 NACK，并决定是否重新入队 ⭐channel.basicNack(deliveryTag, false, true); // 拒绝当前消息，不批量，重新入队System.err.println("💔 消息处理失败，手动发送 NACK 并重新入队 (requeue=true)！Delivery Tag: " + deliveryTag);} catch (Exception nackException) {System.err.println("🔥 发送 NACK 时也发生异常！ Delivery Tag: " + deliveryTag + ", 错误: " + nackException.getMessage());// 这种情况很少见}}System.out.println("---");}}
这个 ManualAckMessageListener 类实现了 ChannelAwareMessageListener 接口，并在 onMessage 方法中直接接收 Message 和 Channel。我们从 message 里提取 deliveryTag，然后用 channel 发送确认信号。模拟失败的逻辑也包含在 onMessage 里。
6. 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);// ⭐ 重要：为了让消费者处理消息和手动确认有时间完成，保持应用运行 ⭐try {System.out.println("\n😴 应用正在运行，消费者正在监听队列。请观察日志中的消息处理和确认结果。");System.out.println(">>> 发送包含 'fail' 的消息会模拟失败并重新入队！");System.out.println(">>> 请手动停止应用以结束演示。");TimeUnit.MINUTES.sleep(5); // 让应用运行一段时间，以便观察效果} catch (InterruptedException e) {Thread.currentThread().interrupt();System.err.println("😴 应用被中断唤醒.");}System.out.println("\n👋 应用演示结束.");}@Overridepublic void run(String... args) throws Exception {System.out.println("🚀 应用启动，开始发送测试消息...");// 发送一条会成功处理的消息messageSender.sendPersistentMessage("This message will succeed!");// 发送一条会先失败几次再成功的消息messageSender.sendPersistentMessage("This message will fail first and then succeed!");// 发送一条会触发 Publisher NACK 的消息 (Exchange 不存在)// messageSender.sendFailedPersistentMessage("This message should be NACKED by publisher confirm!"); // 这条之前演示 Publisher Confirm 用过了，这次主要看消费者确认，可以注释掉System.out.println("\n✅ 所有测试消息已发送提交到 RabbitMQ。请观察消费者端的日志...");}
}
在 Application 类中，我们发送了两条消息，一条会成功处理，另一条会在消费者端模拟失败几次。我们将应用运行时间延长到 5 分钟，以便有充足的时间观察消费者处理消息和 RabbitMQ 重新投递消息的效果。
运行代码并观察结果：
- 确保 RabbitMQ 运行中。
- 运行上面的 Spring Boot 应用。
- 观察控制台日志。
预期结果：
你应该会看到：
- 第一条消息 “This message will succeed!” 正常处理成功并 ACK。
- 第二条消息 “This message will fail first and then succeed!” 会被收到并处理：
- 第一次尝试 (currentAttempt=1) -> 失败 -> NACK 并重新入队。
- 第二次尝试 (currentAttempt=2) -> 失败 -> NACK 并重新入队。
- 第三次尝试 (currentAttempt=3) -> 条件 3 <= 2 不成立，跳过失败模拟 -> 成功处理 -> ACK。
通过这个演示，你就能亲眼看到手动确认模式下，当消费者处理失败时，消息如何被 NACK 并根据 requeue 标志重新回到队列，从而避免消息丢失！👍
消费者确认机制是消息可靠性中防止“已投递未处理”丢失的关键环节，和发送方确认、持久化一起，构成了 RabbitMQ 消息不丢的坚实堡垒！🛡️🏰