import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

@Component
public class MqIdempotentRedisUtil {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 消息冪等占坑
     * @param msgId 消息唯一ID
     * @param expireSeconds 過期時間（大于業(yè)務(wù)最大執(zhí)行時長）
     * @return true=首次消費，false=重復(fù)消費
     */
    public boolean tryLock(String msgId, long expireSeconds) {
        String key = "mq:idempotent:" + msgId;
        // SETNX 原子操作：不存在則設(shè)置，存在則返回false
        return stringRedisTemplate.opsForValue()
                .setIfAbsent(key, "consumed", expireSeconds, TimeUnit.SECONDS);
    }
}

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.io.IOException;

@Component
public class OrderMsgConsumer {

    @Resource
    private MqIdempotentRedisUtil idempotentRedisUtil;

    // 業(yè)務(wù)最大執(zhí)行時長5秒，鎖過期時間設(shè)30秒（預(yù)留緩沖）
    private static final long LOCK_EXPIRE_TIME = 30;

    @RabbitListener(queues = "order.pay.queue")
    public void consume(Message message, Channel channel) throws IOException {
        // 1. 獲取全局唯一消息ID（生產(chǎn)者必須傳遞）
        String msgId = message.getMessageProperties().getHeader("msgId");
        if (msgId == null || "".equals(msgId)) {
            // 無唯一ID，非法消息，直接丟棄
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }

        try {
            // 2. 冪等判斷：占坑失敗=重復(fù)消息
            boolean isFirstConsume = idempotentRedisUtil.tryLock(msgId, LOCK_EXPIRE_TIME);
            if (!isFirstConsume) {
                System.out.println("【重復(fù)消息丟棄】msgId：" + msgId);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                return;
            }

            // 3. 核心業(yè)務(wù)邏輯（下單、支付、積分、物流等）
            doBusiness(msgId);

            // 4. 手動ACK確認消費成功
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            // 消費異常，拒絕消息，重回隊列重試
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            e.printStackTrace();
        }
    }

    private void doBusiness(String msgId) {
        // 模擬業(yè)務(wù)執(zhí)行
        System.out.println("【首次消費成功】處理消息：" + msgId);
    }
}

CREATE TABLE mq_message_record (
    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '主鍵',
    msg_id VARCHAR(64) NOT NULL COMMENT '消息唯一ID',
    business_type VARCHAR(32) COMMENT '業(yè)務(wù)類型',
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE INDEX uk_msg_id (msg_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT 'MQ消息防重表';

import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;

@Service
public class MqIdempotentDbService {

    @Resource
    private MqMessageRecordMapper messageRecordMapper;

    @Transactional(rollbackFor = Exception.class)
    public boolean isFirstConsume(String msgId, String businessType) {
        try {
            // 插入防重記錄
            MqMessageRecord record = new MqMessageRecord();
            record.setMsgId(msgId);
            record.setBusinessType(businessType);
            messageRecordMapper.insert(record);
            return true;
        } catch (DuplicateKeyException e) {
            // 唯一索引沖突，重復(fù)消息
            return false;
        }
    }
}

-- 樂觀鎖更新：僅待支付訂單可更新為已支付
UPDATE order_info 
SET status = 2, pay_time = NOW() 
WHERE order_id = #{orderId} AND status = 1;

@Service
public class OrderService {

    @Resource
    private OrderMapper orderMapper;

    @Transactional(rollbackFor = Exception.class)
    public boolean paySuccess(Long orderId) {
        // 更新行數(shù)=0 說明：訂單已處理，重復(fù)消費
        int rows = orderMapper.updateOrderStatus(orderId, 1, 2);
        return rows > 0;
    }
}