wonder/cs-note

Fork 0

Files

T

hhs 686c0a5bd5 vault backup: 2026-05-24 20:51:06

2026-05-24 20:51:06 +08:00

7.9 KiB

Raw Blame History

tags, create time

MQ 消息幂等性

概述

消息重复是分布式系统的常态，而不是异常。无论是网络超时重试、Consumer Rebalance 还是 Producer 重试，都可能产生重复消息。幂等性保证"执行多次与执行一次效果相同"，是处理重复消息的核心手段。本文介绍四种常见的幂等实现方案，并给出适用场景对比。

正文

为什么会产生重复消息

在讨论幂等方案之前，先搞清楚重复消息的来源。知己知彼，才能对症下药。

网络超时重试：Producer 发送消息后等待 ACK 超时，不确定 Broker 是否收到，于是重试。如果 Broker 其实已经收到了第一条，重试就会产生重复。这是最常见的重复来源。

Consumer Rebalance：Consumer Group 发生 Rebalance 时（如消费者加入或退出），分区会被重新分配。旧 Consumer 可能已经拉取了消息但还没来得及提交 Offset，新 Consumer 又会重新拉取同一批消息。

Producer 重试：Broker 返回一个可重试的错误（如 NotLeaderForPartition），Producer 自动重试发送。如果上一次请求其实已经成功写入，重试就会重复。

[!question] 能不能通过"先查询再处理"的方式避免重复？比如先查数据库，如果订单已存在就跳过？

这种方案存在经典的 TOCTOU（Time of Check to Time of Use） 问题：查询时订单不存在，但在你插入之前，另一个重复消息刚好完成了插入。两个消费者都认为订单不存在，都执行了插入操作。所以"先查询再处理"不能保证幂等，必须借助原子性的去重机制。

幂等的定义

幂等（Idempotent）的数学定义是：f(f(x)) = f(x)。应用到消息处理中，意味着同一条消息被消费一次和被消费 N 次的结果完全一样。

注意幂等不等于"不处理"。幂等是"处理了但效果只生效一次"。比如扣减库存，幂等的做法是记录已处理的消息 ID，重复消息直接跳过扣减逻辑。

方案一：唯一 ID + 去重表

最直观的方案：每条消息携带一个全局唯一的 MessageID，消费前先查去重表，如果已存在则跳过。

Redis Set 实现：将 MessageID 写入 Redis Set，利用 SADD 的原子性实现"检查 + 写入"一步完成。

数据库唯一索引：在去重表上对 MessageID 建唯一索引，插入重复记录时会触发唯一约束冲突，直接跳过。

两种实现各有优劣：Redis 性能高但需要处理过期策略；数据库可靠但吞吐受限于磁盘 I/O。

// 基于 Redis 的消息去重实现
func processWithDedup(rdb *redis.Client, msg *Message) error {
    dedupKey := fmt.Sprintf("mq:dedup:%s", msg.ID)

    // SADD 原子操作：如果 key 不存在则添加并返回 1，已存在则返回 0
    added, err := rdb.SAdd(context.Background(), dedupKey, "1").Result()
    if err != nil {
        return fmt.Errorf("redis sadd: %w", err)
    }

    if added == 0 {
        // 已经处理过，直接跳过
        log.Printf("duplicate message %s, skipped", msg.ID)
        return nil
    }

    // 设置过期时间，避免 Redis 内存无限增长
    rdb.Expire(context.Background(), dedupKey, 24*time.Hour)

    // 执行业务逻辑
    if err := processOrder(msg); err != nil {
        // 业务处理失败，删除去重标记，允许重试
        rdb.Del(context.Background(), dedupKey)
        return err
    }

    return nil
}

这段代码利用了 Redis SADD 的原子性，把"检查是否重复"和"标记已处理"合成一个操作，避免了 TOCTOU 问题。注意业务处理失败时要删除去重标记，否则消息就永远无法被重试了。

方案二：乐观锁（版本号控制）

适用于"更新已有数据"的场景。每条记录带一个版本号，更新时带上版本号条件：

UPDATE inventory SET stock = stock - 1, version = version + 1
WHERE sku = 'A001' AND version = 5;

如果版本号不匹配（说明已经被其他消息更新过），affected_rows 返回 0，业务层判定为重复操作。

这种方案不需要额外的去重表，利用了数据库自身的并发控制能力。适合库存扣减、账户余额变更等场景。

[!question] 乐观锁方案能处理"插入"类操作吗？比如创建订单？

不太适合。乐观锁天然适合"更新"场景，因为更新时已有记录和版本号。对于"插入"类操作，还是用唯一 ID + 去重表更直接。当然你也可以用 INSERT ... ON CONFLICT DO NOTHING（PostgreSQL）来实现插入幂等。

方案三：Token 机制

Token 机制的核心思路是：先获取令牌，再执行操作。

Consumer 处理消息前，先向 Token 服务申请一个全局唯一的 Token。
消费者带着 Token 执行业务操作。
业务服务端校验 Token 是否有效，有效则执行并销毁 Token，无效则拒绝。

Token 服务保证每个 Token 只能使用一次，天然实现了幂等。这种方案适合跨系统的幂等控制，比如下游 API 的幂等调用。

缺点是多了一次网络调用获取 Token，且 Token 服务本身需要高可用。

方案四：状态机约束

很多业务天然具有状态流转特性，利用状态机的单向性可以实现幂等。

比如订单状态：待支付 → 已支付 → 已发货 → 已完成。消费"支付成功"消息时，执行：

UPDATE orders SET status = '已支付'
WHERE order_id = '1001' AND status = '待支付';

如果订单已经是"已支付"状态，affected_rows 为 0，自然幂等。不需要额外的去重表，业务逻辑和幂等校验融为一体。

flowchart TD
    A["收到消息"] --> B["提取消息 MessageID"]
    B --> C["查询去重表"]
    C --> D{"是否已处理?"}
    D -->|"是"| E["跳过,记录日志"]
    D -->|"否"| F["执行业务逻辑"]
    F --> G{"业务执行成功?"}
    G -->|"是"| H["写入去重表"]
    H --> I["提交消费 Offset"]
    G -->|"否"| J["不写去重表"]
    J --> K["等待重试"]

各方案适用场景对比

方案	实现复杂度	性能影响	适用场景	局限性
唯一 ID + 去重表	低	低（Redis）/ 中（DB）	通用场景，最常用	需要额外存储，需处理过期清理
乐观锁	低	低	更新类操作（库存、余额）	不适合插入操作，存在 ABA 问题
Token 机制	高	中（额外网络调用）	跨系统 API 幂等	Token 服务需高可用
状态机约束	低	低	有明确状态流转的业务	业务模型必须支持状态机

实际项目中，这几种方案往往会组合使用。比如：订单系统用状态机约束核心流转，同时用唯一 ID + 去重表作为兜底。没有银弹，关键是理解每种方案的适用边界。

[!question] 如果消息体本身很大，用 Redis Set 存 MessageID 会不会有内存问题？怎么优化？

Redis Set 存的是 MessageID（通常是 36 字符的 UUID），不是消息体本身，所以内存占用可控。但如果消息量极大（每天数亿条），可以考虑以下优化：

Bloom Filter：用极小的内存（每元素几个 bit）判断消息"可能存在"或"一定不存在"。存在误判（false positive），但不会漏判。适合作为第一层快速过滤。
分片 + 过期：按日期分 Key（如 mq:dedup:2026-05-24:{msgID}），过期后自动清理。
数据库归档：Redis 只保留近期去重数据，历史数据归档到数据库。去重窗口通常只需要几天到一周。

7.9 KiB Raw Blame History Unescape Escape