消息幂等消费案例

本文目录导读：

1. 核心概念：什么是幂等性？
2. 经典案例分析与实现方案
3. 总结与最佳实践

我们来深入探讨消息幂等消费,并通过几个经典案例来理解其核心思想与实现方案。

核心概念：什么是幂等性？

幂等性 指的是：一个操作，无论执行一次还是执行多次,其产生的结果都是一样的。

在消息队列（MQ）场景下，幂等消费 指的是：无论消费者从MQ中收到同一条消息多少次，最终对业务系统产生的影响与只消费一次完全相同。

为什么需要幂等性？ 在分布式系统中，消息的“至多一次”（At-Most-Once）、“至少一次”（At-Least-Once）和“精确一次”（Exactly-Once）是三种主要的投递语义。 大部分主流的MQ（如RabbitMQ、Kafka、RocketMQ）为了保证消息不丢失，默认采用“至少一次”的投递语义，这意味着，在极端情况下（如消费者处理成功后、在发送ACK确认之前网络闪断，或消费者在处理过程中发生故障），MQ会重新投递同一条消息，消费者端必须实现幂等,才能安全地处理这些重复消息。

---

经典案例分析与实现方案

下面通过三个典型的业务场景来演示如何实现幂等消费。

数据库唯一约束（最常用、最可靠）

场景： 用户注册、支付订单回调、创建交易流水等，这些业务的核心表都有一个天然的业务唯一键。

问题： 假设用户支付成功后，支付平台回调系统，由于网络抖动，回调消息被发送了两次，如果系统不做处理,就会创建两条相同的订单。

方案： 利用数据库表的唯一约束来防止重复数据。

1. 业务表设计：在关键的业务表中，建立一个或多个字段的唯一索引，订单表中有order_id,这是唯一的业务标识。
2. 消费逻辑：
   • 消费者收到消息后，尝试向业务表中插入一条记录，其中包含业务唯一键（如order_id）。
   • 如果插入成功，说明这是第一次消费,处理后续核心逻辑。
   • 如果插入失败（抛出DuplicateKeyException），说明之前已经消费过这条消息，程序捕获该异常，直接返回“成功”ACK,避免重复处理。

核心代码逻辑示意（伪代码）：

public void onMessage(OrderMessage message) {
    try {
        // 尝试插入，数据库有 order_id 的唯一索引
        orderDao.insert(new Order(message.getOrderId(), message.getAmount())); 
        // 插入成功，继续处理其他业务逻辑，如更新库存、发送通知等
        processBusinessLogic(message); 
    } catch (DuplicateKeyException e) {
        // 插入失败，说明消息重复，静默处理，不报错，直接ACK
        log.info("重复消费消息，已忽略，orderId: {}", message.getOrderId()); 
    }
}

优点： 实现简单，依赖数据库的强一致性，非常可靠。 缺点： 耦合了数据库操作，性能受数据库写入速度限制，如果业务表本身没有合适的唯一键，需要额外建立一张“消费记录表”来使用此方案。

---

Redis SETNX 命令（分布式锁/状态标记）

场景： 用户点击“领取优惠券”、执行定时任务、分布式任务调度等，这些操作的特点是执行次数频繁，且允许有少量的状态存储。

问题： 同一个用户领取同一张优惠券的请求被其他系统重试发送了两次,导致用户账户里多了两张券。

方案： 利用Redis的SETNX（SET if Not eXists）命令作为分布式锁或状态标记。

核心逻辑：

1. 生成唯一Key：将业务的唯一标识作为Redis的Key。coupon:user:123:promo:456（用户123领取活动456的优惠券）。
2. 消费逻辑：
   • 消费者收到消息后，执行 SET key value NX （即SETNX）。
   • 如果返回 true （设置成功），说明这是第一次处理，继续执行核心业务逻辑（发券）。
   • 如果返回 false （Key已存在），说明之前已经处理过，直接ACK,忽略本次消息。
   • 重要： 需要给这个Key设置一个合理的过期时间（如24小时），防止内存无限增长,同时要保证在业务有效期内的幂等性。

核心代码逻辑示意（伪代码）：

public void onMessage(CouponMessage message) {
    // 生成唯一标识 Key
    String lockKey = String.format("coupon:user:%s:promo:%s", 
                                    message.getUserId(), message.getPromoId());
    // 尝试设置锁，值为消息ID，NX表示不存在才设置
    boolean firstTime = redisTemplate.opsForValue()
                            .setIfAbsent(lockKey, message.getMsgId(), 
                                         Duration.ofHours(24));
    if (firstTime) {
        // 第一次，执行业务逻辑
        grantCoupon(message.getUserId(), message.getPromoId());
        log.info("首次消费，发券成功...");
    } else {
        // 重复消费，直接忽略
        log.info("重复消费，已忽略...");
    }
}

优点： 性能好，逻辑简单。 缺点： 依赖Redis的可用性，Key的过期时间需要合理设置，如果业务处理时间超过了Key的过期时间,则可能会发生重复处理。

---

状态机流转（业务层面控制）

场景： 订单状态流转（待支付 -> 已支付 -> 已发货 -> 已完成）、审批流程等,这些业务有明确的状态管理。

问题： “订单已支付”的消息被重复发送，导致订单状态从“已支付”被重复更新,或者出现状态回退。

方案： 在业务对象中维护一个status字段，利用状态机的流转规则来判断。

核心逻辑：

1. 定义状态机：定义合法状态间的流转关系，NOT_PAID 只能流转到 PAID。
2. 消费逻辑：
   • 消费者收到消息后,先查询出当前订单的状态。
   • 判断消息中的目标状态,是否允许从当前状态流转过来。
   • 收到“支付完成”消息：
     • 如果当前状态是 待支付，则更新为 已支付。
     • 如果当前状态已经是 已支付，说明是重复消息,直接忽略。
     • 如果当前状态是 已取消 或 已完成，则可能是数据不一致或乱序，根据策略处理（例如记录日志并拒绝，或抛出异常）。

核心代码逻辑示意（伪代码）：

public void onOrderPaidMessage(OrderMsg msg) {
    Order order = orderDao.findByOrderId(msg.getOrderId());
    // 判断当前状态
    if ("PAID".equals(order.getStatus())) {
        log.info("订单已支付，重复消息，忽略。");
        return;
    }
    // 校验状态机流转规则
    if ("NOT_PAID".equals(order.getStatus())) {
        order.setStatus("PAID");
        orderDao.updateStatus(order);
        log.info("订单状态更新为已支付。");
    } else {
        // 状态异常，告警或特殊处理
        log.error("状态机不允许流转，当前状态: {}", order.getStatus());
    }
}

优点： 从业务逻辑层面天然避免错误状态，逻辑清晰。 缺点： 实现相对复杂，需要提前设计好状态机，如果并发高,需要配合锁来保证状态判断和更新的原子性。

---

总结与最佳实践

方案	优点	缺点	适用场景
数据库唯一约束	最可靠、简单	性能受限（DB写）、需要改造表结构	交易、订单、支付回调、流水记录等核心数据
Redis SETNX	高性能、实现简单	依赖Redis、需处理过期时间	高频发券、领取奖励、接口防重、定时任务
状态机流转	业务逻辑清晰、安全保障高	需要设计状态机、实现复杂	订单状态管理、审批流、有明确生命周期的对象

核心建议：

1. 优先消息自带唯一ID（业务ID）：最好的幂等控制是在消息里带上业务的唯一标识（如orderId、userId+goodsId），而不是依赖MQ生成的messageId，因为同一条业务消息可能因为重发而拥有不同的messageId,但业务标识是唯一的。
2. 配合“消费记录表”：如果业务表过于复杂或没有唯一键，可以在业务库中额外建一张独立的consume_log表，包含消息的唯一业务键和唯一索引,巧妙地将幂等判断转化为简单的数据库插入操作。
3. ACL（Acknowledge）策略：在捕获到重复消费异常（如DuplicateKeyException）时，一定要向MQ返回“成功”确认（ACK），否则MQ会认为消息消费失败，从而无限重发,导致系统压力过大。