RabbitMQ 消息确认机制(ACK)：技术解析与实践指南

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

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

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

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

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

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

channel.basicAck(deliveryTag, multiple);

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

channel.basicNack(deliveryTag, multiple, requeue);

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

channel.basicReject(deliveryTag, requeue);

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

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

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

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

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

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

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto
        retry:
          enabled: true
          initial-interval: 1000    # 第一次重试间隔 1s
          max-interval: 10000       # 最大间隔 10s
          multiplier: 2.0           # 间隔倍增
          max-attempts: 3           # 最大重试次数

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

@Bean
public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
    // 重试耗尽后发送到死信队列
    return new RepublishMessageRecoverer(rabbitTemplate, "dlx.exchange", "dlx.order");
}

@Bean
public Queue orderQueue() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-dead-letter-exchange", "dlx.exchange");       // 死信交换机
    args.put("x-dead-letter-routing-key", "dlx.order");       // 死信路由键
    return new Queue("order.queue", true, false, false, args);
}
@Bean
public Queue deadLetterQueue() {
    return new Queue("order.dlq", true);
}
@Bean
public Binding dlqBinding() {
    return BindingBuilder.bind(deadLetterQueue())
        .to(new DirectExchange("dlx.exchange"))
        .with("dlx.order");
}

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）        消费端确认（消息处理完成）──┘

@Component
@Slf4j
public class OrderMqConsumer {
    @Resource private OrderService orderService;
    @Resource private OrderFailLogRepository failLogRepository;
    @RabbitListener(queues = "${mq.queue.order-process}")
    public void consume(Integer orderId, Channel channel,
                        @Header(AmqpHeaders.DELIVERY_TAG) long tag,
                        @Header(value = "x-death", required = false) List<Map<String, Object>> xDeath) {
        try {
            orderService.processOrder(orderId);
            channel.basicAck(tag, false);
        } catch (RetryableException e) {
            log.warn("订单处理临时失败将重试, orderId={}", orderId, e);
            channel.basicNack(tag, false, true);
        } catch (Exception e) {
            log.error("订单处理不可恢复失败, orderId={}", orderId, e);
            // 记录失败日志，支持运维排查和手动重试
            failLogRepository.save(new OrderFailLog(orderId, e.getMessage()));
            // 拒绝消息，不重回队列（进死信）
            channel.basicReject(tag, false);
        }
    }
}