用户收到两条短信，另一位用户的扣款却“查无此消息”：MQ 重复与丢失的双面案件

周二上午，客服同时甩来两个工单。工单一：“用户说验证码收到了两条，两条内容还不一样，他不知道该填哪个。”工单二：“有个用户被扣了款，但会员权益没到账，我们查了一圈，没找到对应的消息。”

一个是 MQ 多送了一次，一个是 MQ 少送了一次。我盯着这俩工单，感觉像在同一天被同一位供应商“多发货”和“漏发货”，还不能退货。这篇是我梳理 MQ「重复」与「丢失」的排查笔记：先讲清为什么它必然会重复，再讲怎么设计幂等；然后沿着生产端 → Broker → 消费端三个环节查丢消息。以 RocketMQ 为主，兼顾通用性。

一、先接受一个事实：MQ 只承诺“至少一次”

RocketMQ 官方最佳实践里有一句很坦白的话：RocketMQ 无法避免消息重复（Exactly Once），如果业务对重复消费非常敏感，必须在业务层面去重。建议的做法是用消息唯一标识（msgId 或者业务主键）配合数据库校验。

为什么不能保证恰好一次？看一下消息投递的两次“确认”：

• 生产者 → Broker：生产者发消息，Broker 存好后返回结果。如果响应在网络上丢了/超时了，生产者不知道 Broker 到底存没存，只能重发——于是 Broker 里可能有两条一样的消息；
• Broker → 消费者：消费者处理完消息，要向 Broker 提交消费位点（offset）。如果处理完但位点没提交成功（进程被杀、网络断了、重平衡），下次会从旧位点再消费一遍。

分布式系统里，“至少一次 + 幂等处理”才是工程上可落地的组合；追求“精确一次”的代价通常远大于收益。

二、案件一：验证码收到了两条

现象与线索

先拿业务单号去 MQ 里查这条消息。发送短信的消息，我们发送时都设置了 Keys（业务流水号）。用 mqadmin 按 key 查：

$ export NAMESRV=10.0.0.11:9876
$ sh mqadmin queryMsgByKey -n $NAMESRV -t sms-send -k SMS20241015000456
#Message ID                                        #QID  #Offset
0A00030300002A9F000000001F3A0C22                   1     20873011
0A00030300002A9F000000001F3A1A6D                   1     20873012

果然，同一个业务流水号对应两条消息，msgId 不同。这说明重复发生在生产端：同一个业务动作，往 Broker 写了两次。再对照生产者日志：

10:21:08.311 [http-nio-8080-exec-7] INFO  SmsProducer - send sms, key=SMS20241015000456
10:21:11.312 [http-nio-8080-exec-7] WARN  SmsProducer - send timeout, retry once, key=SMS20241015000456
10:21:11.377 [http-nio-8080-exec-7] INFO  SmsProducer - send ok, msgId=0A00030300002A9F000000001F3A1A6D
(此前第一次发送实际已写入 Broker：msgId=0A00030300002A9F000000001F3A0C22)

时间线是这样：第一次发送，Broker 其实已经存好了，但当时 Broker 所在机器有一次短暂的 IO 抖动，响应超过了 3 秒的发送超时时间；客户端认为失败，触发了重试。第二次发送成功。于是 Broker 里有了两条内容相同的消息，消费者各消费一次，用户各收一条短信。（验证码内容不一样，是因为消费者每条消息都生成一次新验证码，进一步说明消费端没有幂等。）

RocketMQ 官方文档对此也有说明：同步发送失败时客户端会重试（默认 retryTimesWhenSendFailed = 2，同步模式下最多重试两次，发送超时默认 3 秒）。这是好意——防丢——但代价就是可能重复。

消费端另外几个“重复”来源

案件一只是一种。我把消费端会重复的情形整理成下面这张图，以后排查对照：

消费端重复的典型来源：

1. 重平衡（Rebalance）：消费者实例上下线（发布、扩缩容、心跳超时）会触发队列重新分配。消费位点是定期异步提交（默认每 5 秒左右）的，所以消费者 A 刚处理完的消息，位点可能还没上报，队列就被分给消费者 B，B 从上次提交的位点开始，这段时间内的消息被再消费一次；
2. 消费超时 / 返回失败：消费逻辑返回 RECONSUME_LATER（或抛异常、超时）时，消息会进入重试队列稍后重投；如果这次失败发生在“副作用已生效，但返回值之前出错”（比如短信已发出，写日志时抛异常），重投就会导致重复副作用；
3. 消费者处理很慢：官方设有消费超时（默认 15 分钟），超时后 Broker 会认为该消息消费失败并重投，而原来的线程可能还在执行；
4. Broker 或消费者异常重启：位点没来得及持久化。

我们这次案件一的根源在生产端重试，但你看，四类来源里任何一类都可能导致“用户收到两条短信”。所以正确的思路不是去逐个消灭重复来源（消灭不完），而是让消费逻辑对重复免疫——幂等。

三、幂等方案：从便宜到严谨

方案 1：数据库唯一键（最推荐，最可靠）

用业务唯一标识建一张“消费记录/去重表”，消费前先插入，插入成功才执行，主键冲突说明处理过：

CREATE TABLE mq_consume_log (
  id           BIGINT PRIMARY KEY AUTO_INCREMENT,
  consumer_grp VARCHAR(64)  NOT NULL,
  biz_key      VARCHAR(128) NOT NULL COMMENT '业务唯一号，如短信流水号；不要用 msgId（重发会变）',
  status       TINYINT      NOT NULL DEFAULT 0 COMMENT '0处理中 1成功',
  create_time  DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
  UNIQUE KEY uk_grp_biz (consumer_grp, biz_key)
) ENGINE=InnoDB;

@Transactional
public void handle(SmsMsg m) {
    try {
        consumeLogMapper.insert(group, m.getBizKey());     // 唯一键冲突 → DuplicateKeyException
    } catch (DuplicateKeyException e) {
        log.info("duplicate message, skip. key={}", m.getBizKey());
        return;                                            // 已处理，直接确认
    }
    doBusiness(m);                                         // 与去重记录同一个本地事务
}

这里有个关键细节：去重键要用业务唯一号，不要用 msgId。案件一里，生产端重试产生的两条消息 msgId 不同、业务号相同——用 msgId 去重，正好挡不住这种重复。

如果 doBusiness 里包含调用外部系统（发短信），数据库事务管不了外部副作用。此时要分两步走：先“抢占”（status=0 处理中），调用成功后置为 1；对状态为处理中且超过阈值的记录，走补偿检查（去运营商查询回执后再决定重发与否）。

方案 2：状态机（天然幂等）

-- 订单只能 待支付 → 已支付，重复消息 update 影响行数为 0
UPDATE orders SET status = 'PAID', pay_time = NOW()
 WHERE order_no = ? AND status = 'WAIT_PAY';
-- 余额类：不要用 “balance = 新值”，也不要盲目 “+ 金额”，要带流水号和条件
INSERT INTO account_flow(flow_no, account_id, amount) VALUES (?, ?, ?);   -- flow_no 唯一
UPDATE account SET balance = balance + ? WHERE id = ?;                     -- 与上一句同一事务

对“状态可以只向前走”的业务，状态机是最省事的幂等：影响行数为 0 就知道已经处理过。

方案 3：Redis SETNX（轻量，但有边界）

// 适合“去重窗口有限”“重复的代价可接受”的场景（如营销短信、埋点上报）
Boolean first = redis.opsForValue().setIfAbsent("mq:dedup:" + bizKey, "1", 24, TimeUnit.HOURS);
if (!Boolean.TRUE.equals(first)) { return; }   // 24 小时内处理过
try {
    doBusiness(m);
} catch (Exception e) {
    redis.delete("mq:dedup:" + bizKey);        // 失败要删掉标记，否则重试永远被跳过
    throw e;
}

Redis 方案的边界要认清：标记先于处理，进程在“标记后、处理前”挂掉，消息重投时会被误判为“已处理”，造成丢失；而 Redis 自身主从切换也可能丢标记造成重复。对资金类业务，不要仅依赖 Redis 去重，数据库唯一键才是最后一道防线。

方案 4：生产端也做幂等

• 发送时设置业务 Keys（message.setKeys(bizNo)），既方便 queryMsgByKey 排查，也是消费者做业务去重的依据；
• 重试次数别乱设大；对 SEND_OK 之外的状态，明确记录并交给补偿，而不是无脑重试；
• 对外的接口（比如用户点“获取验证码”）在入口用限流/防重（同手机号 60 秒内只发一次），从源头减少重复。

四、案件二：一笔扣款没了消息

现象与线索

先确认这条消息到底有没有进 Broker。扣款成功后发“开通会员权益”的消息，业务单号 ORD20241015009911：

$ sh mqadmin queryMsgByKey -n $NAMESRV -t member-grant -k ORD20241015009911
Query Message By Key ... no message found.

Broker 里根本没有这条消息。再看消费端有没有可能“拿到了但处理丢了”：

$ grep 'ORD20241015009911' /data/logs/order/app.log
2024-10-15 09:42:17.201 INFO  OrderService - pay success, order=ORD20241015009911
2024-10-15 09:42:17.230 WARN  MemberProducer - send member-grant message failed, ignore. order=ORD20241015009911
org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout

答案在这里：扣款成功之后发消息超时，代码里 catch 之后只打了一行 WARN 就“ignore”了。看代码：

@Transactional
public void pay(String orderNo) {
    orderMapper.markPaid(orderNo);
    try {
        producer.send(new Message("member-grant", "PAID", orderNo, body));
    } catch (Exception e) {
        log.warn("send member-grant message failed, ignore. order={}", orderNo, e);  // 吞掉异常！
    }
}

这是生产端丢消息的典型：数据库事务已经提交，消息发送失败却被吞掉，二者不一致。而 Broker 也确实存在这样的丢失：如果发送结果不是 SEND_OK，一定要重视。

逐环节排查丢失：生产 → Broker → 消费

生产端：

• 异步发送 / oneway 不看结果：sendOneway 不返回任何结果，网络一抖就丢；异步发送必须实现 SendCallback.onException 并处理；
• 只判断“没抛异常”，不判断 SendStatus：RocketMQ 的 SendResult.sendStatus 有四种：SEND_OK、FLUSH_DISK_TIMEOUT、FLUSH_SLAVE_TIMEOUT、SLAVE_NOT_AVAILABLE。后三种意味着消息已到达 Broker 内存，但刷盘/同步从库没有在规定时间内（默认 5 秒）完成，此时若 Broker 宕机就会丢；
• 异常被吞（案件二）：任何 catch 都必须有落库/重试/告警的后续，不能只打日志；
• 本地事务与消息不一致：先提交事务再发消息，发送失败就丢；先发消息再提交事务，事务回滚消息却已发出，会产生“幽灵消息”。官方方案是事务消息（半消息 + 本地事务 + 回查），或者本地消息表：业务数据和“待发消息”写在同一个本地事务里，由定时任务扫表发送、确认后置状态。

本地消息表的做法：

CREATE TABLE local_msg (
  id        BIGINT PRIMARY KEY AUTO_INCREMENT,
  biz_no    VARCHAR(64) NOT NULL,
  topic     VARCHAR(64) NOT NULL,
  body      TEXT        NOT NULL,
  status    TINYINT     NOT NULL DEFAULT 0 COMMENT '0待发送 1已发送',
  retry_cnt INT         NOT NULL DEFAULT 0,
  next_time DATETIME    NOT NULL,
  UNIQUE KEY uk_biz (biz_no, topic)
);

@Transactional
public void pay(String orderNo) {
    orderMapper.markPaid(orderNo);
    localMsgMapper.insert(orderNo, "member-grant", body);   // 与业务同一事务，要么都成功要么都失败
}
// 定时任务（XXL-JOB，每 5 秒）：扫描 status=0 的记录 → producer.send(..) → SEND_OK 才置为 1，失败则 retry_cnt+1 并退避

Broker 端：看 broker.conf 的两个关键配置——

$ grep -E 'flushDiskType|brokerRole|brokerId|fileReservedTime' /opt/rocketmq/conf/broker.conf
brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH
fileReservedTime=72

默认值是 ASYNC_FLUSH（异步刷盘）和 ASYNC_MASTER（异步复制）。含义：

• ASYNC_FLUSH：消息写入 PageCache 就返回成功，由后台线程稍后刷到磁盘。进程被 kill 不会丢（PageCache 由操作系统保管），但机器断电/内核崩溃会丢尚未刷盘的数据；SYNC_FLUSH 则刷盘后才返回，可靠性最高，吞吐会下降；
• ASYNC_MASTER：主节点写完就返回，异步复制给从节点。主节点磁盘损坏或永久宕机时，尚未复制的消息丢失；SYNC_MASTER 则同步等从节点写成功再返回；
• fileReservedTime：消息保留小时数，默认 72（3 天）。消费者积压超过保留期，未消费消息会被清理，这也是一种“丢”。

要求不丢消息的关键业务，建议 flushDiskType=SYNC_FLUSH + brokerRole=SYNC_MASTER + 至少一个从节点（或使用 DLedger/Raft 多副本方案），并接受吞吐和延迟的代价——这是可靠性与性能的经典取舍：

# broker.conf（要求高可靠的实例）
brokerRole=SYNC_MASTER
flushDiskType=SYNC_FLUSH

消费端：

• 先返回成功，再异步处理：这是最常见的自杀行为。消费回调里把消息丢到本地线程池就直接 return CONSUME_SUCCESS，线程池处理时进程重启，任务丢失，位点却已提交；
• 重试耗尽进入死信后无人处理：%DLQ%消费组名 里躺着的消息不会被自动消费，必须监控和处理；
• 手动重置位点/跳过积压：resetOffsetByTime 之类的操作会让一段时间内的消息不再被消费；
• 订阅关系不一致：同一消费组内不同实例订阅了不同 Tag，会互相覆盖，部分消息被过滤掉——consumerConnection 可以查看订阅（见上一篇堆积排查）。

正确的消费骨架（处理完成后再返回，失败才重试）：

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext ctx) {
    for (MessageExt msg : msgs) {
        try {
            idempotentHandler.handle(parse(msg));     // 幂等 + 同步处理完
        } catch (BizNonRetryableException e) {
            log.warn("non retryable, skip. key={}", msg.getKeys(), e);
        } catch (Exception e) {
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;   // 交给 RocketMQ 重试，超限进死信
        }
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

还可以借助 RocketMQ 的消息轨迹（Message Trace）：开启后可以查看一条消息“谁发的、什么时候到 Broker、谁消费了、耗时多少、成功还是失败”，对排查“到底丢在哪一段”很有用。开启方式是 Broker 侧设置 traceTopicEnable=true，客户端构造 Producer/Consumer 时开启 enableMsgTrace（具体以所用版本的官方文档为准）。

五、止血与修复

止血

1. 重复短信：无法“撤回”，只能对受影响用户做说明；同时对验证码类短信，服务端只认最近一次生成的验证码，别让“收到两条”变成“哪个都错”；
2. 丢失的权益：以数据库为准做对账——找出“已扣款、无权益”的订单，脚本补发权益消息（补发也走幂等）：

-- 对账：已支付但没有权益记录的订单
SELECT o.order_no FROM orders o
LEFT JOIN member_grant g ON g.order_no = o.order_no
WHERE o.status = 'PAID' AND o.pay_time >= '2024-10-15 00:00:00' AND g.order_no IS NULL;

根治

1. 消费端：所有消息处理都套上幂等（唯一键 / 状态机 / 去重表），以业务号为准，不以 msgId 为准；
2. 生产端：关键消息走本地消息表或事务消息，禁止吞异常；检查 SendStatus，非 SEND_OK 进补偿；
3. Broker：关键 Topic 所在集群使用同步刷盘 + 同步复制（或多副本方案），并监控主从延迟；
4. 消费端：先处理后返回，不做“先 ack 再异步”；死信队列监控告警；
5. 每天/每小时做一次对账：以数据库为准，核对“业务表状态”与“下游结果”，兜住所有其他漏洞。

六、复盘

一句话备忘：重复靠幂等（业务号 + 唯一键/状态机），丢失分三段查：生产端看 SendStatus 与被吞的异常，Broker 看刷盘与复制配置，消费端看是不是“先 ack 后处理”；最后用对账兜底。

后来我在团队文档里加了一行醒目的字：“MQ 不欠你一条消息，也不欠你不重复。”同事说这句话很有江湖气。我说不是江湖气，是合同条款——至少一次，就是至少一次，多几次不算违约，少一次才算。而我们要做的，是让“多几次”变成“不痛不痒”。

参考资料

• Apache RocketMQ 4.x：Basic Best Practices——明确“RocketMQ 无法避免消息重复，需业务去重”；SendResult 各状态含义；同步发送的重试与超时；给消息设置 Keys；消费失败返回 RECONSUME_LATER。
• Apache RocketMQ：Consumption Retry——消费失败后的重试间隔、最大次数以及进入死信队列的规则。
• Moment For Technology：RabbitMQ、RocketMQ、Kafka 的事务、丢消息与重复消息策略——三种 MQ 防丢失（同步发送、事务、多副本）与防重复（业务唯一键幂等）的对比。
• 腾讯云开发者社区：常用 RocketMQ 命令——queryMsgByKey、queryMsgById 等 mqadmin 命令用法。