RabbitMQ手动ACK异常：发布者提前收到ACK未监听消费者ACK

你看到的现象是正常的——生产者的ConfirmCallback触发的是MQ服务器的接收确认，和消费者的手动ACK没有任何关系。

拆解一下：

• 生产者的ConfirmCallback：只要MQ服务器成功收到消息、完成路由（如果是持久化队列还要完成持久化），就会立刻给生产者发ACK，这时候消费者可能还没开始处理消息，更别说发ACK了。这个机制的作用是让生产者确认消息已经安全到达MQ，不会丢在半路。
• 消费者的手动ACK：这是消费者告诉MQ“我已经把这条消息处理完了，你可以从队列里删掉它了”，这个操作不会通知生产者。

你的代码里ConfirmCallback工作完全正常，只是你对它的功能理解错了。

下面给两种常用的实现方式：

方式1：同步等待消费者处理结果（RPC模式） #

适合需要同步等待处理结果的场景：

生产者代码修改 #

public void publisherMethod() {
    Thing thing = new Thing("string");
    // 发送消息并等待消费者的应答
    Object response = template.sendAndReceive(exchange, key, thing);
    if (response != null) {
        System.out.println("消费者处理完了，返回消息：" + response);
    }
}

消费者代码修改 #

@RabbitListener(queues = "${rabbitmq.queue.default}", ackMode = "MANUAL")
public String queueDefaultListener(Thing thing, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag)
        throws InterruptedException, IOException {
    System.out.println("Message received");
    Thread.sleep(5000);
    channel.basicAck(tag, false);
    System.out.println("Ack sent");
    // 处理完返回应答消息
    return "消息处理完成";
}

方式2：异步接收处理通知（独立应答队列） #

适合不需要同步等待，只需要异步接收通知的场景：

1. 新增应答队列的配置 #

@Bean
Queue replyQueue() {
    return new Queue("reply-queue");
}

@Bean
DirectExchange replyExchange() {
    return new DirectExchange("reply-exchange");
}

@Bean
Binding replyBinding(Queue replyQueue, DirectExchange replyExchange) {
    return BindingBuilder.bind(replyQueue).to(replyExchange).with("reply-key");
}

2. 生产者添加应答监听器 #

// 监听应答队列，接收消费者的处理完成通知
@RabbitListener(queues = "reply-queue")
public void onReplyReceived(String message) {
    System.out.println("收到消费者通知：" + message);
}

public void publisherMethod() {
    Thing thing = new Thing("string");
    // 可以加个唯一ID，方便关联请求和应答
    CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    template.convertAndSend(exchange, key, thing, correlationData);
    System.out.println("Message sent");
}

3. 消费者处理完发送应答 #

@Autowired
private RabbitTemplate rabbitTemplate;

@RabbitListener(queues = "${rabbitmq.queue.default}", ackMode = "MANUAL")
public void queueDefaultListener(Thing thing, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag)
        throws InterruptedException, IOException {
    System.out.println("Message received");
    Thread.sleep(5000);
    channel.basicAck(tag, false);
    System.out.println("Ack sent");
    // 发送处理完成的通知到应答队列
    rabbitTemplate.convertAndSend("reply-exchange", "reply-key", "消息[" + thing.getContent() + "]处理完成");
}

1. 提前设置ConfirmCallback：你现在是发完消息才设置回调，第一条消息的确认可能抓不到，最好在配置RabbitTemplate的时候就设置：

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
    // 初始化时就绑定确认回调
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        String msgId = correlationData != null ? correlationData.getId() : "无ID";
        if (ack) {
            System.out.println("MQ已接收消息，ID：" + msgId);
        } else {
            System.out.println("MQ接收消息失败，原因：" + cause + "，消息ID：" + msgId);
        }
    });
    return rabbitTemplate;
}

2. 用CorrelationData关联消息：发送消息时带上CorrelationData，能在回调里精准定位是哪条消息的确认结果，排查问题更方便。

内容的提问来源于stack exchange，提问作者StuckOverTheFlow