引言

"用户投诉一笔订单被扣了两次钱，流水截图发过来，两笔扣款间隔 17 秒，金额、商户号、支付单号完全一样。" 查到支付系统时，第一眼以为是第三方支付平台重复回调——对方日志里只回调了一次；再往下查 MQ 监控，发现同一条支付成功消息被投递了两遍，消费者也老老实实执行了两遍扣款，账户服务没有任何去重逻辑。

更讽刺的是，重投递的触发点不是故障，而是消费者处理完业务后没有 ACK：RabbitMQ 等不到确认，30 秒后认为消费者"没处理成功"，把消息重新入队又推了一次。第一次扣款其实已经成功，第二次又成功——幂等性缺失让"至少一次投递"（at-least-once）直接变成了"至少扣一次款"。

这篇文章按事故的真实时间线复盘：消息为什么一定会重复（MQ 的投递语义）→ 重复扣款的代码根因 → 三重保障修复（消费端幂等 + 手动 ACK + 死信队列兜底）→ 唯一索引/Redis SETNX/状态机三种幂等方案的对比选型。代码基于 Spring Boot + Spring AMQP（RabbitMQ），关键语义会对照 RocketMQ 和 Kafka。先给结论：在 MQ 的世界里，"消息只投递一次"是不存在的承诺，"消费逻辑执行多次结果仍然正确"才是工程上唯一可靠的答案——幂等不是优化项，是消费端的生存底线。

一、全景：消息为什么会被重复投递

1.1 三种投递语义

关键点：主流 MQ 承诺的都是"至少一次"，重复投递是设计行为，不是 bug。 很多开发者把重复消费当成异常情况，于是代码里只写"快乐路径"——事故就是这么来的。

1.2 重复发生的五个真实环节

① 生产端重复发送
   发送成功但响应超时/网络抖动 → 生产者重试 → broker 上存了两条相同消息

② 消费端处理成功、ACK 前宕机/重启                              ← 本次事故
   业务已执行完，进程在发 ACK 前被杀 → broker 认为未投递 → 重投

③ 消费端处理超时
   业务执行超过 MQ 的 ack 超时/可见时间（RabbitMQ 30s、
   RocketMQ 消费超时、Kafka max.poll.interval.ms）→ 消息被重新投给另一个消费者

④ 消费端业务异常，触发重试/REQUEUE
   代码 NACK requeue=true，或 RocketMQ 返回 RECONSUME_LATER

⑤ rebalance 重平衡（Kafka/RocketMQ 重平衡、RabbitMQ 队列漂移）
   分区/队列被重新分配，已拉取未提交 offset 的消息被新消费者再消费一次

本次事故命中的是 ② 和 ③ 的组合：扣款逻辑本身耗时（调用银行接口约 8~12 秒），偶发超过消费者配置的 ack 超时，且代码里根本没有手动 ACK，用的是自动 ACK 的默认配置，自动确认的时机又被错误理解——具体细节第二章复盘。

1.3 为什么不能靠 MQ 去重

有人会问：broker 收到相同消息，自己去重不行吗？因为 broker 无法判断两条消息"语义相同"：

• 消息 ID 可能由生产者每次新生成（即使业务内容相同）；
• 就算业务 ID 相同，broker 也分不清这是"重复投递"还是"用户合法的两次相同操作"（比如同一商品确实下了两单）；
• 去重需要保存消费状态，而"消费成功"的判定只有业务代码知道（扣款接口返回成功算不算？扣款成功但记账失败呢？）。

所以行业共识是：broker 负责不丢，消费者负责不重。 幂等的边界必须落在业务侧。

二、事故复盘：17 秒内的两次扣款

2.1 事故时间线

10:00:00  第三方支付平台回调支付网关，支付状态=SUCCESS
10:00:01  网关发送 MQ 消息 PaySuccessMessage(orderNo=P20260928001, amount=199.00)
10:00:01  消费者实例 A 收到消息，开始执行：更新订单状态 → 调用账户扣款
10:00:12  扣款成功（银行接口耗时 11s），订单状态更新完成
10:00:13  ★ 消费者还没来得及 ACK（且实际用的是有缺陷的"处理后自动ack"时序）
          RabbitMQ 认为该消息投递后 30s 未确认？——实际触发原因见 2.3
10:00:18  同一条消息被重新投递给消费者实例 B
10:00:18  B 再次执行：更新订单状态（UPDATE 影响 0 行，代码没检查）→ 又调一次扣款
10:00:29  第二次扣款成功。用户被扣两次，间隔 17 秒

2.2 漏洞代码

@Component
@Slf4j
public class PaySuccessListener {

    @Autowired private AccountService accountService;
    @Autowired private OrderService orderService;

    // ✗ 问题一：NONE/AUTO 模式下 ACK 语义不清，且业务非幂等
    @RabbitListener(queues = "q.pay.success")
    public void onMessage(PaySuccessMessage msg) {
        log.info("收到支付成功消息: {}", msg.getOrderNo());

        // ✗ 问题二：订单状态更新结果不检查（重复执行时返回0但继续往下走）
        orderService.markPaid(msg.getOrderNo());

        // ✗ 问题三：扣款无任何幂等保护，同一个 orderNo 可以扣无数次
        accountService.deduct(msg.getUserId(), msg.getAmount(), msg.getOrderNo());
    }
}

// ✗ 扣款 Service：没有唯一约束、没有状态判断、没有去重
public void deduct(Long userId, BigDecimal amount, String bizNo) {
    Account account = accountMapper.selectByUserId(userId);
    account.setBalance(account.getBalance().subtract(amount));
    accountMapper.updateById(account);

    // 流水直接插入，同一 bizNo 可以插多条
    AccountFlow flow = new AccountFlow();
    flow.setUserId(userId);
    flow.setAmount(amount.negate());
    flow.setBizNo(bizNo);
    flow.setType("PAY_DEDUCT");
    accountFlowMapper.insert(flow);
}

三处缺陷叠加：MQ 层可能重投、Listener 层不去重、Service 层和数据库层都没有唯一约束。任何一层有防护，事故都不会发生——这是典型的"纵深防御全面失守"。

2.3 ACK 到底哪里出了问题

排查组最初争论的焦点是"自动 ACK 为什么没生效"。Spring AMQP 的 ACK 模式有三个：

事故实例的真实配置是 AUTO，问题出在两点：

1. 扣款链路中一个非关键步骤（发站内信）偶发抛异常，导致整个监听方法异常退出 → 容器按 AUTO 语义 NACK 且 defaultRequeueRejected=true（默认）→ 消息立刻 REQUEUE 重投。而此时扣款已经完成。
2. 账户服务对银行接口的调用没有把"订单状态已更新"作为前置条件——markPaid 的 UPDATE 即使因状态已是 PAID 而影响 0 行，代码也没中断，扣款照样执行。

// ✗ markPaid 的 SQL 没有任何防护，重复调用不报错
@Update("UPDATE t_order SET status='PAID', pay_time=NOW() WHERE order_no=#{orderNo}")
int markPaid(@Param("orderNo") String orderNo);
// Service 里连返回值都没接收，更没判断

AUTO 模式的本质问题：它保证的是"消息处理流程"的重试，而不是"业务结果"的一次生效。 方法里有多个副作用时，异常点之前的副作用已经落库，重试必然从头再来——不幂等就重复。

2.4 排查过程中的关键证据

# ① RabbitMQ 管理界面：消息的 Redelivered 标记
#    Queues → q.pay.success → Get messages → 第二次消息红字段 "redelivered: true"

# ② 应用日志：同一 orderNo 两条"收到支付成功消息"，且投递到了不同实例
grep "P20260928001" app-instance-a.log app-instance-b.log
app-instance-a.log:10:00:01 收到支付成功消息: P20260928001  consumerQueue=1 redelivered=false
app-instance-b.log:10:00:18 收到支付成功消息: P20260928001  consumerQueue=0 redelivered=true  ← 重投递

# ③ 账户流水表：同一 biz_no 两条扣款流水
SELECT id, user_id, amount, biz_no, create_time
FROM t_account_flow WHERE biz_no = 'P20260928001';
# 1001 | 10086 | -199.00 | P20260928001 | 10:00:12
# 1002 | 10086 | -199.00 | P20260928001 | 10:00:29   ← 重复扣款铁证

证据链闭合：broker 重投递（redelivered=true）+ 消费端无去重 + 流水表无唯一约束。修复也必须三层一起补。

三、第一重保障：消费端幂等（核心）

幂等的通用思路只有一句话：为每次业务操作分配一个全局唯一的业务标识，执行前先判断"这个标识是否已经处理过"，处理过就直接返回成功，没处理过才执行。 支付场景里这个标识天然就是支付单号 orderNo（或支付流水号），绝不能用 messageId（每次投递可能不同，且生产端重试会变）。

下面给三种落地方案，从数据库兜底到缓存加速再到业务状态机。

3.1 方案一：数据库唯一索引（最可靠的兜底）

给账户流水表的业务单号加唯一索引：

-- 业务流水号唯一：同一笔业务永远只能有一条流水
ALTER TABLE t_account_flow
    ADD UNIQUE INDEX uk_biz_no_type (biz_no, type);

消费代码利用 DuplicateKeyException 识别重复：

@Transactional(rollbackFor = Exception.class)
public void deduct(Long userId, BigDecimal amount, String bizNo) {
    AccountFlow flow = buildFlow(userId, amount, bizNo);
    try {
        accountFlowMapper.insert(flow);
    } catch (DuplicateKeyException e) {
        // 唯一索引冲突 = 这笔业务已经扣过款，幂等返回，不报错（否则会触发无意义的 MQ 重试）
        log.warn("扣款流水已存在，幂等跳过: bizNo={}", bizNo);
        return;
    }

    // 余额扣减用"条件 UPDATE"原子完成，不要先查后算（防并发超扣）
    int rows = accountMapper.deductWithCheck(userId, amount.negate());
    if (rows == 0) {
        // 余额不足：这是业务异常，不能吞——抛出后走 MQ 重试/死信流程
        throw new BizException(ErrorCode.BALANCE_NOT_ENOUGH);
    }
}

<!-- 余额扣减：带余额条件的原子 UPDATE，并发安全 -->
<update id="deductWithCheck">
    UPDATE t_account
    SET balance = balance + #{delta}
    WHERE user_id = #{userId}
      AND balance + #{delta} >= 0
</update>

优点：不依赖任何中间件，数据库约束是最后的铁壁，即使应用层全部失效也不可能插入两条流水。缺点：重复消息会打到数据库产生异常（量大时是额外负载），且只适用于"有流水表/插入型"的业务。

注意事务边界：唯一索引冲突后如果直接 return，方法上的 @Transactional 在 Spring 默认配置下不会因为捕获了异常就回滚（异常被吞了），这正是我们想要的——插入失败的那条记录本来就不存在。但如果同一事务里还有其他写操作要回滚，需要手动标记 rollback。本方案把 insert 放第一步、冲突即 return，语义最干净。

3.2 方案二：Redis SETNX（高并发首选，挡住绝大多数重复）

支付回调高峰 QPS 可能上千，让每条重复消息都打一次数据库不划算。用 Redis 的 SET key value NX EX seconds 做第一道快速去重，判断+占位一步完成：

@Component
@Slf4j
public class IdempotentService {

    private final StringRedisTemplate redisTemplate;

    /** 处理中/已完成的标记保留 24h，覆盖 MQ 最大重试周期 + 业务追溯窗口 */
    private static final long TTL_HOURS = 24;

    public IdempotentService(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * 尝试占用幂等键。
     * @return true=首次，可以执行业务；false=重复消息，应直接 ACK 跳过
     */
    public boolean tryAcquire(String key) {
        // ★ 必须是原子的 setIfAbsent（SET NX EX），禁止先 exists 再 set——两步之间有并发窗口
        Boolean ok = redisTemplate.opsForValue()
                .setIfAbsent(key, "1", TTL_HOURS, TimeUnit.HOURS);
        boolean acquired = Boolean.TRUE.equals(ok);
        log.info("幂等检查 key={} result={}", key, acquired ? "FIRST" : "DUPLICATE");
        return acquired;
    }

    /** 业务执行失败时释放键，让 MQ 重试能够重新进入业务逻辑 */
    public void release(String key) {
        redisTemplate.delete(key);
        log.info("业务失败，释放幂等键 key={}", key);
    }
}

消费者改造：

@Component
@Slf4j
public class PaySuccessListener {

    private static final String IDEMPOTENT_PREFIX = "mq:consumed:pay_success:";

    @RabbitListener(queues = "q.pay.success")
    public void onMessage(PaySuccessMessage msg, Channel channel,
                          @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        String key = IDEMPOTENT_PREFIX + msg.getOrderNo();   // ★ 用业务单号，不用 messageId
        boolean firstTime;
        try {
            firstTime = idempotentService.tryAcquire(key);
        } catch (Exception e) {
            // Redis 异常的降级策略见 3.4，这里先按 Fail-Closed 处理：拒绝并稍后重试
            log.error("Redis 不可用，拒绝消息等待重试: orderNo={}", msg.getOrderNo(), e);
            channel.basicNack(tag, false, true);   // requeue=true
            return;
        }

        if (!firstTime) {
            log.warn("重复消息，ACK 跳过: orderNo={}", msg.getOrderNo());
            channel.basicAck(tag, false);          // 重复消息必须 ACK，否则会无限重投
            return;
        }

        try {
            payService.handlePaySuccess(msg);      // 内部仍要有 3.1 的唯一索引兜底
            channel.basicAck(tag, false);          // ★ 业务成功后才 ACK
        } catch (Exception e) {
            idempotentService.release(key);        // ★ 失败释放键，保证重试可重新执行
            log.error("消费失败，NACK 进入重试: orderNo={}", msg.getOrderNo(), e);
            channel.basicNack(tag, false, false);  // requeue=false → 进死信队列（见第五章）
        }
    }
}

三个用血泪换来的细节：

1. 禁止 exists + set 两步走。高并发下两条重复消息可能同时通过 exists 检查再各自 set，幂等形同虚设。必须 SET NX 原子占位。
2. 业务失败必须释放幂等键。否则会出现"第一次因银行抖动失败，但键已占住，后续所有重试都被判为重复跳过，这笔支付永远处理不了"的永久性误拦截。
3. Redis 去重不能替代数据库唯一约束。Redis 键过期、被淘汰（maxmemory 驱逐）、主从切换丢数据都可能让去重失效——它是"加速带"，唯一索引才是"防撞梁"。支付链路两者必须同时上。

3.3 方案三：业务状态机（语义最完整）

支付/订单类业务天然有状态流转，状态本身就是幂等依据：

INIT(待支付) → PAID(已支付) → SETTLED(已结算)
                  ↘ REFUNDED(已退款)

用条件 UPDATE 做状态机的原子守卫，只有从前置状态流转成功才继续执行后续动作：

@Transactional(rollbackFor = Exception.class)
public void handlePaySuccess(PaySuccessMessage msg) {
    // ★ 条件更新：只有 INIT 状态才能变成 PAID，返回 0 说明已经处理过
    int rows = orderMapper.casStatus(msg.getOrderNo(),
            OrderStatus.INIT.name(), OrderStatus.PAID.name());

    if (rows == 0) {
        Order order = orderMapper.selectByOrderNo(msg.getOrderNo());
        if (order != null && OrderStatus.PAID.name().equals(order.getStatus())) {
            log.warn("订单已是支付成功状态，幂等返回: orderNo={}", msg.getOrderNo());
            return;   // 标准的重复消息，静默成功（让消费者 ACK）
        }
        // 其他状态（如已退款后又收到迟到的回调）属于异常分支，按业务规则告警/入死信
        throw new BizException(ErrorCode.ILLEGAL_ORDER_STATUS);
    }

    // 状态抢占成功，才执行扣款等后续副作用
    accountService.deduct(msg.getUserId(), msg.getAmount(), msg.getOrderNo());
    notifyService.sendPaySuccessNotice(msg.getOrderNo());
}

<update id="casStatus">
    UPDATE t_order
    SET status = #{toStatus}, update_time = NOW()
    WHERE order_no = #{orderNo} AND status = #{fromStatus}
</update>

这其实是轻量级的乐观锁：数据库行锁保证同一时刻只有一个线程能把状态从 INIT 改成 PAID，其他并发重复消息全部 rows=0。它比单纯唯一索引更贴近业务语义——不仅防重复，还防止了状态的非法跳转（比如已退款的单又被扣费）。

3.4 Redis 不可用时的降级抉择

Redis 去重引入了新的依赖，必须回答"Redis 挂了怎么办"。两种策略没有绝对对错，取决于下游是否有兜底：

支付场景的正确答案是 Fail-Closed + 数据库唯一索引双保险：Redis 正常时挡住重复流量，Redis 异常时暂停消费等待恢复，哪怕消息堆积也绝不能放重复扣款过去；同时唯一索引保证极端情况下数据库也不会出现两条流水。做成配置项，给不同业务不同的选择：

@ConfigurationProperties(prefix = "mq.idempotent")
@Data
public class IdempotentProperties {
    /** CLOSED：Redis故障时拒绝并重试；OPEN：放行（要求下游强幂等） */
    private String redisFallback = "CLOSED";
    private long ttlHours = 24;
}

3.5 三种方案对比与组合选型

实战组合（支付系统修复后的最终形态）：

消息进入
  → Redis SETNX 快速去重（挡住 99.9% 重复，保护数据库）
  → 订单状态机 CAS（INIT→PAID，语义级守卫，防并发+非法跳转）
  → 扣款 + 流水插入（uk_biz_no_type 唯一索引，最后一道铁壁）

三层各司其职：Redis 管性能，状态机管业务正确性，唯一索引管数据底线。

四、第二重保障：手动 ACK——让"处理成功"和"确认"严格对齐

4.1 为什么支付类消费必须用 MANUAL ACK

第三章代码已经切到了手动 ACK，这里讲清配置和语义。Spring AMQP 配置：

spring:
  rabbitmq:
    host: rabbitmq-prod
    listener:
      simple:
        acknowledge-mode: manual          # ★ 手动确认
        prefetch: 10                        # 每个消费者预取数量，支付类建议调小(1~10)防堆积压垮
        default-requeue-rejected: false     # ★ 业务异常 NACK 后不重回原队列，直接进死信
        retry:
          enabled: true
          max-attempts: 3                   # 容器本地重试次数（针对可恢复异常）
          initial-interval: 1000
          multiplier: 2

手动 ACK 的四条分支必须在代码里写全，这是排查时最容易漏的地方：

4.2 幂等键与 ACK 的时序（最关键的一张图）

拉到消息
                 │
        SETNX 占位幂等键
          │              │
       成功(首次)       失败(重复)
          │              │
      执行业务        basicAck 丢弃
          │
     ┌────┴────┐
   成功       失败
     │          │
 basicAck   释放幂等键
  (键保留)   basicNack(requeue=false)
                 │
             死信队列（TTL延迟后重投，重新走全流程）

核心纪律：键的占位在业务之前，键的释放在业务失败之后，ACK 永远在业务结果确定之后。 顺序反了就会出现两类事故：先 ACK 后业务 → 业务失败消息丢了；业务失败不释放键 → 消息重试但永远被幂等拦截。

4.3 消费侧的异常分类

不是所有异常都该重试，无差别重试会制造重复风暴：

try {
    payService.handlePaySuccess(msg);
    channel.basicAck(tag, false);
} catch (BizException e) {
    idempotentService.release(key);
    if (e.getCode() == ErrorCode.BALANCE_NOT_ENOUGH) {
        // 余额不足：重试可能有用（等用户充值/补偿到账），进死信延迟队列
        channel.basicNack(tag, false, false);
    } else {
        // 非法状态等业务错误：重试一万次也不会成功，直接进死信人工处理
        log.error("业务不可恢复异常，转死信人工处理: orderNo={}", msg.getOrderNo(), e);
        channel.basicNack(tag, false, false);
    }
} catch (Exception e) {
    idempotentService.release(key);
    // 未知异常（数据库/银行连接）：进死信，由死信的延迟机制控制重试节奏
    log.error("消费未知异常: orderNo={}", msg.getOrderNo(), e);
    channel.basicNack(tag, false, false);
}

4.4 RocketMQ / Kafka 的对照

无论哪种 MQ，消费端业务幂等的代码写法完全一致，换的只是确认 API。RocketMQ 用户特别注意：默认 16 次重试意味着你的扣款逻辑可能在没有幂等保护时被执行 16 次，比 RabbitMQ 默认行为更"执着"。

五、第三重保障：死信队列兜底——失败消息的隔离区

5.1 为什么需要死信队列

没有死信队列时，消费失败的消息只有两个选择：requeue 无限重试（把消费者拖死、日志刷爆），或者直接丢弃（支付消息丢了就是资损）。死信队列（DLQ）提供第三条路：暂时无法处理的消息移到隔离队列，不阻塞主队列，由专门的机制延迟重试或人工处理。

5.2 RabbitMQ DLX + TTL 延迟重试落地

架构：

q.pay.success（主队列）
   │ NACK(requeue=false)
   ↓
DLX 交换机 ex.pay.dlx
   ↓
q.pay.dlx.wait（死信暂存队列，设 TTL，到期后路由到恢复交换机）
   │ TTL 到期（如 30s/1min/5min 递增，用多个队列模拟）
   ↓
ex.pay.retry → q.pay.success（重新消费）
                          ↓ 超过最大次数
                     q.pay.dlx.terminal（人工处理队列 + 告警）

声明代码：

@Configuration
public class PayQueueConfig {

    public static final String Q_PAY_SUCCESS = "q.pay.success";
    public static final String EX_PAY_DLX = "ex.pay.dlx";
    public static final String Q_PAY_WAIT_30S = "q.pay.dlx.wait.30s";
    public static final String EX_PAY_RETRY = "ex.pay.retry";
    public static final String Q_PAY_TERMINAL = "q.pay.dlx.terminal";

    /** 业务主队列：绑定死信交换机 */
    @Bean
    public Queue paySuccessQueue() {
        return QueueBuilder.durable(Q_PAY_SUCCESS)
                .withArgument("x-dead-letter-exchange", EX_PAY_DLX)
                .withArgument("x-dead-letter-routing-key", "wait30s")
                .build();
    }

    /** 死信等待队列：30 秒 TTL，到期后消息成为 ex.pay.retry 的死信，再被送回主队列 */
    @Bean
    public Queue wait30sQueue() {
        return QueueBuilder.durable(Q_PAY_WAIT_30S)
                .withArgument("x-message-ttl", 30000)
                .withArgument("x-dead-letter-exchange", EX_PAY_RETRY)
                .withArgument("x-dead-letter-routing-key", "back")
                .build();
    }

    /** 终态死信队列：超过重试次数的消息进这里，配套告警和人工后台 */
    @Bean
    public Queue terminalQueue() {
        return QueueBuilder.durable(Q_PAY_TERMINAL).build();
    }

    @Bean
    DirectExchange payDlxExchange() { return new DirectExchange(EX_PAY_DLX, true, false); }

    @Bean
    DirectExchange payRetryExchange() { return new DirectExchange(EX_PAY_RETRY, true, false); }

    @Bean
    Binding dlxBinding() {
        return BindingBuilder.bind(wait30sQueue()).to(payDlxExchange()).with("wait30s");
    }

    @Bean
    Binding retryBinding() {
        // 回到主队列重新消费
        return BindingBuilder.bind(paySuccessQueue()).to(payRetryExchange()).with("back");
    }

    @Bean
    Binding terminalBinding() {
        return BindingBuilder.bind(terminalQueue()).to(payDlxExchange()).with("terminal");
    }
}

重试次数控制放在消息头（x-death 是 RabbitMQ 自动记录的死信历史，也可以自定义 header 计数）：

// 进死信前根据已重试次数决定：继续延迟重试 or 进终态队列
Long retried = (Long) msgProps.getHeaders().getOrDefault("x-retry-count", 0L);
if (retried >= 5) {
    // 转人工：发到 terminal 路由，并触发告警
    rabbitTemplate.convertAndSend(EX_PAY_DLX, "terminal", message, m -> {
        m.getMessageProperties().setHeader("x-retry-count", retried + 1);
        return m;
    });
    channel.basicAck(tag, false);
    alertService.sendP1("支付消息重试5次仍失败，需人工处理: " + msg.getOrderNo());
}

5.3 死信队列的运维纪律

1. 死信队列必须有消费者监控：没有监控的死信队列等于黑洞。队列深度 > 0 持续 5 分钟即告警，深度 > 100 触发 P1。
2. 重试要有上限和退避：30s → 1min → 5min 递增，避免故障期疯狂重试把银行接口打挂。
3. 终态消息要有人工处理界面：展示消息体、失败原因、重试次数，支持"修复数据后一键重放"。
4. 重放消息走原队列，不要直接改数据库——保证完整的业务校验和流水再次经过幂等三层防护。

5.4 事故后的补救：重复扣款怎么回滚

技术防护之外，资金事故还要有数据修复 SOP。当晚排查组的操作顺序：

-- ① 冻结涉事账户的自动扣费，止损
-- ② 精确圈定重复流水：同 biz_no+type 出现多条且金额相同的非首条
SELECT user_id, biz_no, COUNT(*) c
FROM t_account_flow
WHERE type = 'PAY_DEDUCT' AND create_time BETWEEN '2026-09-28 09:00' AND '2026-09-28 11:00'
GROUP BY user_id, biz_no
HAVING c > 1;

-- ③ 对账：与银行侧实际扣款笔数核对（银行只扣了一次的，说明第二次是内部重复记账）
-- ④ 红冲：对重复流水做反向补偿流水（不能物理删除，资金流水必须可审计）
INSERT INTO t_account_flow (user_id, amount, biz_no, type, remark)
SELECT user_id, -amount, CONCAT(biz_no, '-REVERSE'), 'REVERSE_DEDUCT', '重复扣款冲正'
FROM t_account_flow WHERE id = #{duplicateFlowId};
-- ⑤ 余额加回、用户通知、事后复盘

这个 SOP 后来固化成了财务对账系统的自动巡检任务：每日扫描 biz_no 重复流水，发现即告警。

六、生产级消费端的完整模板

把三重保障合到一个可直接抄的模板里：

@Component
@Slf4j
public class AbstractIdempotentListener {

    /** 子类提供：幂等键前缀、业务处理、最大重试次数 */
    protected void consume(Channel channel, long tag, String bizKey,
                           String idempotentKey, int maxRetry,
                           Runnable business) throws IOException {
        boolean acquired;
        try {
            acquired = idempotentService.tryAcquire(idempotentKey);
        } catch (Exception e) {
            log.error("幂等组件异常 Fail-Closed，NACK重投: key={}", idempotentKey, e);
            channel.basicNack(tag, false, true);
            return;
        }

        if (!acquired) {
            log.warn("重复消费，ACK跳过: bizKey={}", bizKey);
            channel.basicAck(tag, false);
            return;
        }

        try {
            business.run();
            channel.basicAck(tag, false);
        } catch (RetryableException e) {
            idempotentService.release(idempotentKey);
            log.warn("可恢复异常，转死信延迟重试: bizKey={}", bizKey, e);
            channel.basicNack(tag, false, false);   // DLX + TTL
        } catch (Exception e) {
            idempotentService.release(idempotentKey);
            log.error("不可恢复异常，ACK并转终态死信: bizKey={}", bizKey, e);
            // 实际项目中此处把消息转发 terminal 队列后再 ACK
            channel.basicNack(tag, false, false);
        }
    }
}

配置纪律清单：

•  幂等键一律用业务唯一单号，不用 messageId/uuid
•  SETNX 原子占位 + TTL 覆盖最大重试周期（建议 24h）
•  业务失败释放键，业务成功保留键
•  手动 ACK，ACK 在业务结果确定之后
•  default-requeue-rejected=false，失败消息统一进 DLX，不原地无限重投
•  数据库唯一索引/状态机作为最终兜底，Redis 不可用时 Fail-Closed
•  死信队列深度监控 + 重试上限 + 人工重放入口
•  五类日志：幂等键内容/TTL、FIRST 或 DUPLICATE、命中跳过原因、降级原因、键清理结果

七、常见问题

7.1 用消息自带的 messageId 做幂等键不行吗？

不行，这是最高频的误用。messageId 在三个环节会变：生产端重试时 SDK 可能重新生成；不同 MQ 中间件的 messageId 生成规则不同；RocketMQ 的 msgId 是客户端生成的、也可能与服务端 offsetId 混淆。幂等键必须来自业务语义上唯一的标识——支付单号、订单号、业务流水号。一条判断标准：用户针对同一业务动作的所有消息（包括重发），这个值都相同；不同业务动作即使内容相似，这个值也不同。

7.2 幂等通过了但业务只执行了一半（扣款成功、记账失败）怎么办？

这是"部分成功"问题，幂等解决不了，需要的是本地事务/事务消息/对账兜底。原则：一个消费者方法内的多个写操作放进同一个数据库事务，一起成功或一起回滚；跨系统的副作用（扣款 + 发券 + 通知）用本地消息表或 Saga 拆成可补偿步骤，失败可重试、可回滚；资金系统永远保留 T+1 对账作为最终校验。幂等保证"重复执行无害"，事务保证"单次执行完整"，两者是互补关系。

7.3 Redis SETNX 的键设多久合适？过期了消息又重投会不会重复？

TTL 必须覆盖"MQ 最大重试周期 + 业务处理窗口"。RocketMQ 默认 16 次重试、跨度可达数小时，键至少保留 24h；RabbitMQ 死信重试 5 次以内、跨度几十分钟，24h 也足够。极端情况下键过期与重投同时发生，仍有数据库唯一索引/状态机兜底——这也是不能把 Redis 当唯一防线的原因。如果业务需要永久防重（财务凭证类），去重记录应落数据库表而不是只靠 Redis TTL。

7.4 消费者处理很慢导致超时重投，除了幂等还能怎么治？

幂等是"正确性"兜底，超时治理是"性能"手段，两者都要：① 调大消费处理超时阈值（RabbitMQ 的 ack 没有 broker 硬超时，但要关注消费者失联检测；RocketMQ consumeTimeout 默认 15 分钟；Kafka max.poll.interval.ms 默认 5 分钟）；② 缩小 prefetch/批量大小，避免一次拉太多处理不完；③ 慢操作异步化——监听方法只做落库，银行调用交给下游异步编排；④ 耗时操作设置合理的调用超时和熔断，别让一个下游拖垮消费节奏。

7.5 生产端重复发送（消息进了 broker 两条）怎么防？

消费端幂等天然能吸收生产端重复（两条消息业务键相同，第二条被幂等跳过），所以消费端做好了这个问题自动消失。如果要在生产端也做加固：发送时用业务单号作为消息去重键（RocketMQ 的 keys、RabbitMQ 的 messageId 自定义赋值），发送失败重试时保持同一业务键；更严格的场景用 RocketMQ 事务消息或本地消息表保证"本地事务与发消息"的一致性。但记住：生产端手段不能替代消费端幂等，因为重投还可能发生在 broker/消费环节。

7.6 幂等拦截会不会把合法消息误杀？

会，主要有两种误杀：① 业务失败但幂等键没释放（修复方案：失败必释放）；② Redis 键被错误地设置成了非业务唯一维度（比如用 userId 而不是 orderNo，导致同用户第二笔正常支付被拦）。防范手段：键的维度必须精确到"业务动作"，并记录 3.5 节要求的五类日志；上线前用测试用例覆盖"重复消息只执行一次"和"合法新消息不被拦截"两个方向；监控幂等命中率，DUPLICATE 比例异常飙升（比如超过 30%）时要排查是否键维度设计错误或上游故障。

八、总结

排查与修复速查卡

【怎么确认是重复消费】
  RabbitMQ：Get messages 看 redelivered=true；日志按业务单号搜出多条消费记录
  RocketMQ：控制台消息轨迹看 reconsumeTimes > 0
  Kafka：消费日志同 key/offset 重复；rebalance 时间点与重复时间吻合
  数据库：业务流水表同 biz_no 多条记录（事后铁证）

【三重保障】
  ① 消费端幂等（业务正确性）
     Redis SETNX 原子占位（挡流量）→ 状态机 CAS（管语义）→ 唯一索引（守底线）
     键=业务单号，TTL≥重试周期；失败释放、成功保留
  ② 手动 ACK（投递对齐）
     acknowledge-mode=manual；成功/重复→Ack；可恢复失败→释放键+NACK进DLX
     default-requeue-rejected=false，禁止原地无限重投
  ③ 死信队列（失败兜底）
     主队列 → DLX → TTL等待队列 → 回主队列（退避重试，上限5次）
     → 终态队列 + 监控告警 + 人工重放

【方案选型】
  资金/订单：三列都上（Redis + 状态机 + 唯一索引），Redis故障 Fail-Closed
  普通通知/日志：唯一索引或Redis去重即可
  天然状态流转业务：状态机CAS为主，Redis加速
  查询/读类消费：Redis去重窗口即可（重复查询无副作用，要求低）

给团队的建议

一句话

消息重复消费不是小概率事故，而是"至少一次"投递语义下必然会发生的正常事件——生产者重试、ACK 前宕机、消费超时、requeue、rebalance 每一个环节都会让同一条消息再次出现，broker 按设计就不承诺去重，所以幂等的责任只能落在业务侧。这次重复扣款的根因是三层防线全部失守：消费者用 AUTO 确认让"站内信失败"触发了整条消息 NACK 重投，Listener 对同 orderNo 不做任何判断，Service 层先查后扣没有并发保护，流水表连唯一索引都没有。修复的三重保障缺一不可：第一重消费端幂等是核心——Redis SETNX 必须原子占位（禁止 exists+set 两步）、键用业务单号而非 messageId、业务失败必须释放键避免永久误拦截，状态机 CAS 用条件 UPDATE 守住语义正确性，数据库唯一索引作为不依赖任何中间件的最终铁壁，支付链路在 Redis 故障时选择 Fail-Closed 而不是拿资金安全赌可用性；第二重手动 ACK 让确认严格发生在业务结果确定之后，成功与重复都 ACK、可恢复失败释放幂等键后 NACK 进死信，关闭 default-requeue-rejected 杜绝原地无限重投；第三重死信队列用 DLX+TTL 做退避重试、设上限、终态消息进人工队列并配深度告警。记住"Redis 管性能、状态机管业务、唯一索引管底线、ACK 管投递、死信管失败、对账管资损"这六道分工，消息队列里 90% 的重复消费类资损事故都可以避免。

互动话题：你们系统里出现过消息重复消费吗？最后是靠唯一索引、Redis 去重还是状态机发现的？有没有踩过"业务失败没释放幂等键导致消息永远跳过"或者"AUTO ACK 下非核心步骤异常导致整条重投"的坑？评论区聊聊你的幂等设计。

参考资料

• RabbitMQ 官方文档：Consumer Acknowledgements and Publisher Confirms
• RabbitMQ 官方文档：Dead Letter Exchanges
• Spring AMQP 参考文档：Receiving Messages（acknowledge-mode）
• RocketMQ 官方文档：消费者重试与死信主题
• Apache Kafka 官方文档：Message Delivery Semantics（at-least-once）
• Redis 官方文档：SET 命令（NX/EX 原子语义）
• Spring Boot 优雅实现接口幂等（同系列文章）