消息重复消费导致订单超卖？这4个绝招让消息只被消费一次！

作为一名后端开发，经历过太多因为消息重复消费导致的"惨案"：

• 某电商平台因为订单消息重复消费，导致同一件商品被卖出了10次，老板差点把我祭天
• 某支付系统因为回调消息重复处理，用户的钱被扣了3次，客服电话被打爆
• 某积分系统因为MQ消息重复投递，用户的积分被重复扣减，用户直接投诉到消协

消息重复消费，可以说是分布式系统中最让人头疼的问题之一。今天就结合自己踩过的坑，跟大家聊聊如何确保消息只被消费一次，让你的系统稳如老狗。

一、消息为什么会重复消费？这3个场景你肯定遇到过

在讲解决方案之前，我们先得搞清楚消息为什么会重复。根据我多年的踩坑经验，主要有以下3个原因：

1. 网络抖动导致的重复投递

MQ为了保证消息不丢失，都有重试机制。当网络出现抖动时，生产者以为消息没发出去，就会重新发送，导致消费者收到多条相同的消息。

真实场景：我之前用RabbitMQ的时候，网络偶尔会出现500ms的延迟，生产者以为消息没发出去，结果重复发送了3次，消费者一下子收到了3条一模一样的订单消息。

2. 消费者处理超时导致的重复投递

当消费者处理消息时间过长，超过了MQ的超时时间，MQ会认为消息处理失败，就会重新投递这个消息。

真实场景：我们有个订单处理逻辑，需要调用多个外部接口，有一次外部接口响应特别慢，处理时间超过了30秒，RocketMQ以为处理失败了，又投递了一次，结果同一个订单被处理了两次。

3. 消费者重启导致的重复消费

当消费者重启时，如果使用的是自动ack模式，可能会导致之前已经消费但未确认的消息被重新消费。

真实场景：某次上线更新，我们重启了订单服务，结果重启前已经处理过的100多条订单消息又被重新消费了一遍，直接导致库存超卖。

二、4个核心技巧，让消息只被消费一次

针对以上问题，我总结了4个实战中验证有效的解决方案。

1. 幂等性设计：从根源上解决问题

原理：让消息的处理具备幂等性，即多次处理同一条消息的结果是一样的。

实现方法：

• 使用唯一标识：每条消息都有一个全局唯一的消息ID，处理前先检查这个ID是否已经被处理过
• 业务幂等：利用业务本身的幂等特性，比如"支付"这个操作本身就是幂等的

实战代码：

// 基于Redis的幂等处理
public class MessageProcessor {
    private RedisTemplate<String, String> redisTemplate;
    
    public void processMessage(String messageId, String message) {
        // 使用SETNX命令，如果key不存在则设置成功，返回1；如果已存在则设置失败，返回0
        Boolean result = redisTemplate.opsForValue()
            .setIfAbsent("msg_processed:" + messageId, "1", Duration.ofHours(1));
        
        if (Boolean.TRUE.equals(result)) {
            // 消息未处理过，执行业务逻辑
            doBusinessLogic(message);
        } else {
            // 消息已处理过，直接返回
            log.info("消息{}已处理过，跳过", messageId);
        }
    }
}

实战案例：我之前负责的支付系统，每条支付消息都有一个全局唯一的transactionId。处理前先检查这个transactionId是否已经在数据库中存在，如果存在就直接返回，不存在就处理。这样即使有重复消息，也不会重复扣款。

2. 消息去重表：数据库级别的保障

原理：在数据库中建立一个消息去重表，记录已经处理过的消息ID。

表结构设计：

CREATE TABLE message_dedup (
    message_id VARCHAR(64) PRIMARY KEY,
    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_create_time (create_time)
);

处理流程：

1. 收到消息后，先尝试向去重表插入消息ID
2. 如果插入成功，说明消息未处理过，执行业务逻辑
3. 如果插入失败（主键冲突），说明消息已处理过，直接跳过

实战案例：某电商平台的订单系统，每天处理上百万条订单消息。我们专门建了一个message_dedup表来记录处理过的消息ID，用数据库的唯一索引来保证幂等性。运行3年多，从未出现过重复消费的问题。

3. Redis分布式锁：高并发场景下的利器

原理：利用Redis的分布式锁机制，确保同一时间只有一个实例在处理某条消息。

实现代码：

public class DistributedLockMessageProcessor {
    private RedisTemplate<String, String> redisTemplate;
    
    public void processMessage(String messageId, String message) {
        String lockKey = "lock:msg:" + messageId;
        String lockValue = UUID.randomUUID().toString();
        
        try {
            // 获取分布式锁，设置10秒过期时间
            Boolean locked = redisTemplate.opsForValue()
                .setIfAbsent(lockKey, lockValue, Duration.ofSeconds(10));
            
            if (Boolean.TRUE.equals(locked)) {
                // 获取锁成功，处理消息
                doBusinessLogic(message);
                
                // 处理完成后删除幂等key
                redisTemplate.delete("msg_processed:" + messageId);
            } else {
                // 获取锁失败，说明其他实例正在处理
                log.info("消息{}正在被其他实例处理", messageId);
            }
        } finally {
            // 释放锁，使用Lua脚本保证原子性
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                          "return redis.call('del', KEYS[1]) else return 0 end";
            redisTemplate.execute(new DefaultRedisScript<>(script, Long.class),
                Collections.singletonList(lockKey), lockValue);
        }
    }
}

实战案例：我们有个秒杀系统，QPS能达到5万。为了避免同一个订单被重复处理，我们用Redis分布式锁来保证每个订单消息只被一个实例处理。配合幂等性设计，双重保险，秒杀活动从未出过问题。

4. 状态机+乐观锁：复杂业务场景的终极方案

原理：利用数据库的乐观锁机制，通过版本号或状态机来保证消息处理的幂等性。

订单状态机示例：

public class OrderService {
    
    public void handlePaymentMessage(String orderId) {
        // 使用乐观锁更新订单状态
        int affectedRows = orderMapper.updateOrderStatus(
            orderId, 
            "PAID",     // 目标状态
            "UNPAID"    // 期望的当前状态
        );
        
        if (affectedRows == 1) {
            // 更新成功，说明订单状态从UNPAID变为PAID
            // 执行后续业务：扣减库存、发送通知等
            doAfterPayment(orderId);
        } else {
            // 更新失败，说明订单状态已经不是UNPAID
            log.info("订单{}状态已变更，跳过处理", orderId);
        }
    }
}

数据库SQL：

UPDATE orders 
SET status = 'PAID', update_time = NOW() 
WHERE order_id = #{orderId} AND status = 'UNPAID';

实战案例：某物流系统的订单状态更新，我们设计了完整的状态机：待发货→已发货→已签收→已完成。每个状态转换都有对应的SQL条件，确保只有当前状态正确才能进行下一步操作。即使有重复消息，也不会导致状态错乱。

三、实战案例：某电商平台的订单消息幂等处理全过程

下面我分享一个真实的电商平台订单消息幂等处理案例，看看我们是如何一步步实施的。

1. 项目背景

老系统使用RabbitMQ处理订单消息，经常出现消息重复消费导致的订单重复创建问题。新系统需要支持高并发订单处理，同时保证消息幂等性。

2. 技术方案

我们采用了"三重保障"的策略：

第一层：消息去重

• 每条消息都有全局唯一的messageId
• 使用Redis的SETNX命令进行幂等判断
• 设置合理的过期时间（1小时）

第二层：业务幂等

• 订单号全局唯一，数据库有唯一索引
• 订单状态机设计，防止重复更新
• 支付流水号唯一，防止重复支付

第三层：补偿机制

• 定时任务扫描异常订单
• 人工干预处理特殊情况
• 完整的监控和告警机制

3. 核心代码实现

@Service
public class OrderMessageProcessor {
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    @Autowired
    private OrderService orderService;
    
    @RabbitListener(queues = "order_queue")
    public void processOrderMessage(Message message) {
        String messageId = message.getMessageProperties().getMessageId();
        String orderData = new String(message.getBody());
        
        try {
            // 第一重保障：消息幂等
            if (!checkMessageProcessed(messageId)) {
                // 第二重保障：业务幂等
                Order order = JSON.parseObject(orderData, Order.class);
                orderService.createOrderWithIdempotency(order);
                
                // 标记消息已处理
                markMessageProcessed(messageId);
            }
            
            // 手动ack确认消息
            message.getMessageProperties().getChannel()
                .basicAck(message.getMessageProperties().getDeliveryTag(), false);
            
        } catch (Exception e) {
            log.error("处理订单消息失败，messageId: {}", messageId, e);
            // 消息重试机制
            throw new AmqpRejectAndDontRequeueException("处理失败，等待重试");
        }
    }
    
    private boolean checkMessageProcessed(String messageId) {
        return Boolean.TRUE.equals(
            redisTemplate.hasKey("order_msg_processed:" + messageId)
        );
    }
    
    private void markMessageProcessed(String messageId) {
        redisTemplate.opsForValue()
            .set("order_msg_processed:" + messageId, "1", Duration.ofHours(1));
    }
}

4. 效果验证

• 消息重复率：从原来的5%降低到0.01%
• 订单准确率：达到99.99%，基本没有重复订单
• 系统性能：支持10万QPS的订单处理，响应时间<100ms

四、5个避坑小贴士

最后，我再分享几个消息幂等处理中的关键细节：

1. 消息ID要全局唯一：使用UUID或者业务ID+时间戳的组合，确保不会重复
2. 幂等key要设置过期时间：避免Redis内存无限增长，一般设置1-24小时
3. 异常处理要完善：不要让异常消息无限重试，设置最大重试次数
4. 监控要到位：监控消息重复率、处理成功率等关键指标
5. 测试要充分：上线前要进行大量并发测试，模拟各种异常情况

五、总结

消息重复消费是分布式系统中不可避免的问题，但只要掌握了正确的幂等处理方法，完全可以避免业务影响。记住：幂等性设计不是锦上添花，而是分布式系统的必修课。

如果你觉得这篇文章对你有帮助，欢迎点赞、在看、转发三连！关注「服务端技术精选」，获取更多技术干货。