深入浅出 RabbitMQ-RabbitMQ消息确认机制（ACK）

spring:
  rabbitmq:
    host: 你的RabbitMQ地址
    port: 5672
    username: 用户名
    password: 密码
    # 消费者监听配置：开启手动确认
    listener:
      simple:
        acknowledge-mode: manual  # 手动确认模式
        prefetch: 1  # 可选：每次只拉取1条消息，处理完再拉取下一条，避免Unacked消息堆积

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
@RabbitListener(queues = "coupon.release.queue")  // 监听的队列名
public class CouponReleaseConsumer {

    // 处理消息的核心方法：body是消息内容，message是消息对象，channel是通信信道
    @RabbitHandler
    public void releaseCouponRecord(String body, Message message, Channel channel) throws IOException {
        // 1. 获取当前消息的DeliveryTag（关键：唯一标识当前消息）
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        System.out.println("当前消息投递序号（DeliveryTag）：" + deliveryTag);
        System.out.println("消息内容：" + body);
        System.out.println("消息完整信息：" + message.toString());

        try {
            // 2. 核心业务逻辑：比如“释放用户优惠券”
            processCouponRelease(body);  // 自定义业务方法，如操作数据库、调用接口
            
            // 3. 业务处理成功：手动发送ACK，告知RabbitMQ删除消息
            // 参数1：deliveryTag：当前消息的投递序号
            // 参数2：multiple：是否“批量确认”（false=仅确认当前消息，true=确认当前及之前所有未确认的消息）
            channel.basicAck(deliveryTag, false);
            System.out.println("消息处理成功，已发送ACK，RabbitMQ将删除该消息");

        } catch (Exception e) {
            // 4. 业务处理失败：根据场景决定“重新入队”还是“拒绝入队”
            System.err.println("消息处理失败，body：" + body + "，错误信息：" + e.getMessage());
            
            // 先获取当前消息的重试次数（自定义header，记录重试次数）
            Integer retryCount = message.getMessageProperties().getHeader("retryCount");
            if (retryCount == null) retryCount = 0;

            if (retryCount < 3) {  // 重试阈值：最多重试3次
                // 重试：拒绝消息但允许重新入队，同时更新重试次数
                message.getMessageProperties().setHeader("retryCount", retryCount + 1);
                // 参数1：deliveryTag：当前消息序号
                // 参数2：multiple：是否批量拒绝
                // 参数3：requeue：是否重新入队（true=放回队列，等待再次消费；false=拒绝入队）
                channel.basicNack(deliveryTag, false, true);
                System.out.println("重试次数：" + (retryCount + 1) + "，消息已重新入队");

            } else {
                // 超过重试阈值：拒绝入队，同时记录异常消息到数据库（方便人工审核）
                channel.basicNack(deliveryTag, false, false);
                saveFailedMessage(body, e.getMessage());  // 自定义方法：将异常消息存库
                System.out.println("超过重试阈值（3次），消息已拒绝入队，已记录到异常表");
            }
        }
    }

    // 模拟：业务处理方法（释放优惠券）
    private void processCouponRelease(String body) {
        // 这里写实际业务逻辑：比如解析body中的用户ID、优惠券ID，更新数据库状态等
        // if (数据库连接失败) throw new RuntimeException("数据库异常");
    }

    // 模拟：异常消息记录方法
    private void saveFailedMessage(String body, String errorMsg) {
        // 实际开发中：将body（消息内容）、errorMsg（错误信息）、时间戳等存入数据库
        // 示例SQL：insert into failed_message (content, error_msg, create_time) values (?, ?, now())
    }
}