SpringBoot防止消息重复消费的方法步骤

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 过期时间（大于业务最大执行时长）
     * @return true=首次消费，false=重复消费
     */
    public boolean tryLock(String msgId, long expireSeconds) {
        String key = "mq:idempotent:" + msgId;
        // SETNX 原子操作：不存在则设置，存在则返回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;

    // 业务最大执行时长5秒，锁过期时间设30秒（预留缓冲）
    private static final long LOCK_EXPIRE_TIME = 30;

    @RabbitListener(queues = "order.pay.queue")
    public void consume(Message message, Channel channel) throws IOException {
        // 1. 获取全局唯一消息ID（生产者必须传递）
        String msgId = message.getMessageProperties().getHeader("msgId");
        if (msgId == null || "".equals(msgId)) {
            // 无唯一ID，非法消息，直接丢弃
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }

        try {
            // 2. 幂等判断：占坑失败=重复消息
            boolean isFirstConsume = idempotentRedisUtil.tryLock(msgId, LOCK_EXPIRE_TIME);
            if (!isFirstConsume) {
                System.out.println("【重复消息丢弃】msgId：" + msgId);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                return;
            }

            // 3. 核心业务逻辑（下单、支付、积分、物流等）
            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) {
        // 模拟业务执行
        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 '业务类型',
    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) {
            // 唯一索引冲突，重复消息
            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) {
        // 更新行数=0 说明：订单已处理，重复消费
        int rows = orderMapper.updateOrderStatus(orderId, 1, 2);
        return rows > 0;
    }
}