RabbitMQ 消息确认机制（ACK）：技术解析与实践

一、ACK 是什么

ACK（Acknowledge）是消费者告诉 Broker “这条消息我处理完了，你可以删除了” 的信号。没有 ACK，Broker 不知道消息是否被成功处理，也就无法决定是否删除消息。

Producer → Broker(Queue) → Consumer │ ├─ 处理成功 → ACK → Broker 删除消息 │ ├─ 处理失败 → NACK/Reject → Broker 重新入队或进死信 │ └─ 消费者宕机 → 超时无 ACK → Broker 重新分发给其他消费者

注：

博客：

https://blog.csdn.net/badao_liumang_qizhi

二、为什么需要 ACK

没有 ACK 机制会怎样：

ACK 机制的核心保证：消息至少被成功处理一次（At Least Once）。

三、三种确认模式

1. 自动确认（auto）

spring:rabbitmq:listener:simple:acknowledge-mode:auto

行为：

• 消息推送到消费者方法后，Spring 根据方法执行结果自动决定：
  • 方法正常返回 → 自动 ACK
  • 方法抛异常 → 自动 NACK + requeue

@RabbitListener(queues="order.queue")publicvoidconsume(IntegerorderId){orderService.process(orderId);// 正常返回 → 框架自动 ACK// 抛异常 → 框架自动 NACK，消息重新入队}

优点：代码简单，不需要手动管理缺点：异常时无限 requeue 可能导致消息循环（需配合重试策略）

2. 手动确认（manual）

spring:rabbitmq:listener:simple:acknowledge-mode:manual

行为：

• 消费者必须在代码中显式调用basicAck或basicNack
• 不调用则消息一直处于 Unacked 状态，消费者断连后消息重新入队

@RabbitListener(queues="order.queue")publicvoidconsume(IntegerorderId,Channelchannel,@Header(AmqpHeaders.DELIVERY_TAG)longdeliveryTag){try{orderService.process(orderId);// 处理成功，确认消息channel.basicAck(deliveryTag,false);}catch(BusinessExceptione){// 业务异常，不重试，直接丢弃（或进死信）channel.basicReject(deliveryTag,false);}catch(Exceptione){// 临时异常，重新入队等待重试channel.basicNack(deliveryTag,false,true);}}

优点：精细控制，可以区分不同异常做不同处理缺点：代码复杂度增加，忘记 ACK 会导致消息堆积

3. 无确认（none）

spring:rabbitmq:listener:simple:acknowledge-mode:none

行为：

• Broker 推送后立即删除消息，不等待消费者确认
• 等同于"发后即忘"

优点：最高吞吐缺点：消息可能丢失，仅适用于可丢失的场景（如日志采集、监控指标）

四、ACK 相关的三个操作

basicAck — 确认成功

channel.basicAck(deliveryTag,multiple);

// 逐条确认channel.basicAck(deliveryTag,false);// 批量确认（确认 tag<=5 的所有消息）channel.basicAck(5,true);

basicNack — 否定确认（批量）

channel.basicNack(deliveryTag,multiple,requeue);

// 单条否定，重新入队channel.basicNack(deliveryTag,false,true);// 单条否定，不重回队列（进入死信队列或丢弃）channel.basicNack(deliveryTag,false,false);

basicReject — 否定确认（单条）

channel.basicReject(deliveryTag,requeue);

功能与basicNack相同，但只能操作单条消息（没有 multiple 参数）。

五、deliveryTag 详解

Channel 1: msg_a(tag=1) msg_b(tag=2) msg_c(tag=3) Channel 2: msg_x(tag=1) msg_y(tag=2)

• deliveryTag 是每个 Channel 独立递增的序号
• 消费者收到消息时携带 tag，ACK 时回传这个 tag
• Broker 通过 (channel + tag) 唯一定位一条消息

六、Unacked 消息与 prefetch 的关系

prefetch = 3 时： Broker Queue: [msg4] [msg5] [msg6] ... ↑ 等待中，需要消费者 ACK 释放额度 Consumer 手中: [msg1:处理中] [msg2:处理中] [msg3:处理中] ↑ Unacked 数量 = 3 = prefetch 上限

• Broker 最多推prefetch条未确认消息给消费者
• 消费者 ACK 一条后，Broker 才会推下一条
• 如果消费者始终不 ACK：达到 prefetch 上限后不再推送新消息

消费者宕机时：

• Broker 检测到 Channel/Connection 断开
• 所有 Unacked 消息自动回到 Queue 头部
• 其他消费者可以重新消费这些消息

七、常见问题与陷阱

陷阱1：忘记 ACK 导致消息堆积

// 错误示例：manual 模式下忘记 ACK@RabbitListener(queues="order.queue")publicvoidconsume(IntegerorderId,Channelchannel,@Header(AmqpHeaders.DELIVERY_TAG)longtag){orderService.process(orderId);// 忘记调用 channel.basicAck(tag, false);// 后果：消息一直 Unacked，达到 prefetch 后不再推送新消息}

排查方式：RabbitMQ 管理后台看 Queue 的 Unacked 数量持续增长。

陷阱2：异常时 requeue 导致无限循环

// 错误示例：所有异常都 requeue@RabbitListener(queues="order.queue")publicvoidconsume(IntegerorderId,Channelchannel,@Header(AmqpHeaders.DELIVERY_TAG)longtag){try{orderService.process(orderId);channel.basicAck(tag,false);}catch(Exceptione){// 如果是参数错误（永远不会成功），每次 requeue 都会再失败channel.basicNack(tag,false,true);// 无限循环！}}

正确做法：区分可重试异常和不可重试异常：

try{orderService.process(orderId);channel.basicAck(tag,false);}catch(RetryableExceptione){// 临时故障（网络超时、锁冲突），重新入队channel.basicNack(tag,false,true);}catch(Exceptione){// 不可恢复（参数错误、数据不存在），拒绝不重回log.error("消费失败且不可重试, orderId={}",orderId,e);channel.basicReject(tag,false);// 进死信或丢弃}

陷阱3：批量确认丢消息

// 风险示例：multiple=true 批量确认channel.basicAck(tag,true);// 如果 tag=5，则 tag 1~5 全部确认// 如果 tag=3 的消息其实还没处理完，也会被确认掉

建议：除非有明确的批量处理逻辑，否则始终用multiple=false。

八、重试策略配置

Spring 内置重试（auto 模式下）

spring:rabbitmq:listener:simple:acknowledge-mode:autoretry:enabled:trueinitial-interval:1000# 第一次重试间隔 1smax-interval:10000# 最大间隔 10smultiplier:2.0# 间隔倍增max-attempts:3# 最大重试次数

流程：

第1次消费失败 → 等1s → 第2次重试 → 等2s → 第3次重试 → 仍失败 → 进入 MessageRecoverer

自定义失败处理器

@BeanpublicMessageRecoverermessageRecoverer(RabbitTemplaterabbitTemplate){// 重试耗尽后发送到死信队列returnnewRepublishMessageRecoverer(rabbitTemplate,"dlx.exchange","dlx.order");}

死信队列（DLX）配置

@BeanpublicQueueorderQueue(){Map<String,Object>args=newHashMap<>();args.put("x-dead-letter-exchange","dlx.exchange");// 死信交换机args.put("x-dead-letter-routing-key","dlx.order");// 死信路由键returnnewQueue("order.queue",true,false,false,args);}@BeanpublicQueuedeadLetterQueue(){returnnewQueue("order.dlq",true);}@BeanpublicBindingdlqBinding(){returnBindingBuilder.bind(deadLetterQueue()).to(newDirectExchange("dlx.exchange")).with("dlx.order");}

消息进入死信队列的条件：

• 被 reject/nack 且 requeue=false
• 消息 TTL 过期
• 队列达到最大长度

九、生产者确认（Publisher Confirm）

ACK 不仅消费端有，生产端也有——确认消息成功到达 Broker：

spring:rabbitmq:publisher-confirm-type:correlated# 异步确认publisher-returns:true# 路由失败回调@Component public class OrderMqProducer{@Resource private RabbitTemplate rabbitTemplate; @PostConstruct public void init(){// 消息到达 Exchange 的确认 rabbitTemplate.setConfirmCallback((data,ack,cause)->{if (!ack){log.error("消息未到达Exchange,cause={}",cause); // 重发或记录}}); // 消息无法路由到 Queue 的回调 rabbitTemplate.setReturnsCallback(returned->{log.error("消息无法路由,exchange={},routingKey={},replyText={}",returned.getExchange(),returned.getRoutingKey(),returned.getReplyText());});}}

完整的消息可靠性链路：

Producer → Confirm → Exchange → Routing → Queue → Consumer → ACK │ │ └── 生产端确认（消息到达 Broker） 消费端确认（消息处理完成）──┘

十、三种确认模式选型指南

实际项目建议：

• 默认用auto+ 重试配置 + 死信队列，覆盖 90% 场景
• 核心资金类业务用manual，精确控制每条消息的命运
• none仅用于明确标注"允许丢失"的非关键数据

十一、完整示例：可靠消费模板

@Component@Slf4jpublicclassOrderMqConsumer{@ResourceprivateOrderServiceorderService;@ResourceprivateOrderFailLogRepositoryfailLogRepository;@RabbitListener(queues="${mq.queue.order-process}")publicvoidconsume(IntegerorderId,Channelchannel,@Header(AmqpHeaders.DELIVERY_TAG)longtag,@Header(value="x-death",required=false)List<Map<String,Object>>xDeath){try{orderService.processOrder(orderId);channel.basicAck(tag,false);}catch(RetryableExceptione){log.warn("订单处理临时失败将重试, orderId={}",orderId,e);channel.basicNack(tag,false,true);}catch(Exceptione){log.error("订单处理不可恢复失败, orderId={}",orderId,e);// 记录失败日志，支持运维排查和手动重试failLogRepository.save(newOrderFailLog(orderId,e.getMessage()));// 拒绝消息，不重回队列（进死信）channel.basicReject(tag,false);}}}

十二、总结