高并发消息与异步：消息幂等与去重

可视化演示

消息幂等与去重保障消费者在网络重试或消息重复投递时不会产生重复副作用。介绍生产者序列号、消费端去重表与Redi... 切换模式查看不同维度。

重复消息成因

快递驿站为什么总是收到重复件？不是快递小哥手滑，而是整个配送链路都抱着“宁可多送，不可漏送”的信念。网络超时了，重试；签收确认丢了，重试；送完一单发现电脑没记录，再补送一单。消息系统也一样，重复不是偶然事故，而是“至少一次投递”语义下的默认行为。

在消息队列中，生产者发送消息到中转站（Broker），如果超时没有收到确认，会重新发送；中转站把消息交给消费者后，如果消费者没有及时返回确认，中转站会再次投递；消费者处理完业务、还没提交消费位置时就重启了，重启后会从上次位置重新拉取。于是同一条消息可以走三条“回头路”，从不同环节再次出现在你面前。

三次重试不是简单的三选一，而是可能叠加发生。假设网络重试概率是 \(p_1\)，中转站重投概率是 \(p_2\)，消费者重拉概率是 \(p_3\)，那么消息至少被重复投递一次的概率如下：

所以，聪明的架构师不会天真地追求“杜绝重复”，而是让下游消费者在重复到来时“纹丝不动”。正如驿站即使收到三件同样的包裹，也只签收一次、入库一次。

幂等核心概念

驿站签收台有一个朴素原则：签收章盖一次，和盖一百次，库存系统里都只有一行“已签收”。这就是幂等。数学上也是如此：一个操作执行若干次，最终结果和执行一次完全相同。

消息消费中的幂等，就是重复消息不会产生重复副作用。比如支付回调只入账一次，扣减库存只扣一次，发放优惠券只发一张。反观“查询订单状态”天然幂等，查多少次答案都一样；但“给用户余额加 100 元”不是幂等，加两次就是 200 元。所以不能把所有消息都同等对待，必须识别哪些操作存在副作用。

幂等的目标不是让代码毫无变化，而是让系统状态收敛到同一个最终结果。用函数表达就是：

回到驿站：在包裹上盖一个“已签收”章是幂等的，因为章印不会叠加成两个；但拆开包裹取出商品的动作不幂等，拆两次就会多拿一件。因此，驿站必须用台账把“已签收”这件事记录下来，让拆包动作只发生一次。

生产者序列号

驿站每天收到几百个包裹，怎么区分“同一个包裹”和“两个长得一样的包裹”？答案是运单号。每一个包裹在发货时就印上一个唯一编号，后续无论重试、补送还是换人送，运单号都不会变。消息系统里的“运单号”就是生产者序列号。

生产者在发送消息前，会在消息头或消息体内放入一个全局唯一 ID。最简单的做法是“业务主键 + 事件类型 + 自增序号”，例如订单号 20260612001、事件类型 CREATE、序号 01，拼起来就是 20260612001-CREATE-01。也可以使用 UUID、雪花算法等全局 ID 生成器。重点是：这个 ID 标识的是“同一次业务事件”，而不是“同一条网络包”。

消费者拿到消息后，第一件事就是取出这个序列号，作为去重判定的钥匙。如果没有这把钥匙，后面所有去重方案都无法落地。构造序列号的方式可以写成：

运单号解决的是“身份问题”。驿站可以不知道包裹里装的是什么，但只要运单号相同，就必须按同一个包裹处理。生产者序列号是整套幂等体系的起点，也是后续去重表的索引钥匙。

消费端去重表

驿站门口放着一本厚厚的签收台账，每收到一个包裹，管理员先翻一翻：运单号已经登记过，就签收栏画个勾并请快递员把货带走；没登记过，才写一行新记录。这个台账在消息系统里就是一张“消费端去重表”。

去重表的结构很简单：以 msg_id 作为主键或唯一索引，再加一个写入时间。消费者收到消息后，先尝试插入去重表。如果插入成功，说明这是第一次见到这条消息，可以继续执行业务；如果插入时发生唯一键冲突，说明之前已经处理过，直接返回成功即可。

这里最关键的一点是：去重操作和业务操作必须在同一个数据库事务里。如果先写去重表、再写业务表，去重表提交了但业务失败，重试时就会被错误地当作“已处理”跳过。正确的做法是：开启事务 → 插入去重记录 → 执行业务写入 → 一起提交；任何一步失败则一起回滚。结果可以表达为：

有了这本台账，重复包裹即使来一百次，也永远只会在第一行记录里被“签收”一次。这就是消费端去重表能在生产环境中直接消灭重复扣款的原因。

Redis去重方案

生意变好后，驿站门口排起了长队。每一单都去数据库翻台账，翻一次就要几十毫秒，高峰期根本扛不住。于是管理员在大门口加了一台“高速盖章机”：机器只有一个章，谁先来谁先盖，后到的包裹发现章已经盖过，直接拒收。这台机器就是 Redis 的 SETNX 命令。

SETNX 的意思是“只有当 key 不存在时才能设置成功”。第一个消费者执行 SET dedup:msgId 1 NX 会返回成功，第二个消费者再用同一个 key 执行就会被拒绝。在 Redis 中，这个操作是原子性的，多个并发消费者同时到达，也只有一个能成功。如果担心 key 永远占用内存，可以加上 EX ttl 设置过期时间。于是完整的去重命令变成：

高速盖章机虽然快，但也不是万能的。Redis 重启或 key 过期后，同一消息可能再次通过。所以在资金安全等强一致场景，通常用 Redis 做短时间高频去重，再用数据库去重表做最终兜底，形成“双保险”。

边界与权衡

选择哪种去重方案，就像选择驿站的规模：小驿站一个手写本就能记账，大型转运中心必须上高速盖章机，但机器贵、会断电、也会丢数据。没有绝对最优方案，只有针对不同业务诉求的取舍。

三种方案在三个维度上差异明显。生产者序列号是“免费的前提”，只负责给消息一个身份，不承担拦截职责；消费端去重表拥有最强一致性，能把去重和业务绑在同一个事务里，但每来一条消息就多一次数据库写入，高并发下成本高；Redis 去重延迟低、吞吐高，但受限于 TTL 窗口，且不具备事务回滚能力。

去重的边界也随之变化。数据库去重表可以做到永久精确，适合支付、订单等核心链路；Redis 去重窗口通常只覆盖秒级到天级，适合验证码、限流、活动发券等短时间重复攻击。在消息语义上，绝大多数系统接受“至少一次投递 + 幂等消费”的组合，从效果上无限逼近“恰好一次处理”。整体成本可以用读写开销的叠加来评估：

所以，面对“重复消息”这道题，真正要回答的不是“哪个方案最好”，而是“业务能为重复窗口忍受多大的代价”。愿意多花数据库成本，就换来永久精确；追求极致性能，就必须接受 Redis 的短窗口和弱持久化。

生产实践复盘

纸上谈兵再多，不如看一次真实事故。某支付系统曾出现用户被重复扣款，监控显示同一笔支付回调被消费了两次。查到最后，根因是消费者把支付结果写入业务库后，还没来得及提交消费位置就宕机重启了，消息队列重新投递了同一回调；更糟的是，当时去重表与业务表不在同一事务里，去重记录先提交，业务更新却失败了，后续重试全部被当作“已处理”跳过。

修复方案分三步：第一，为支付回调生成全局唯一的 msg_id，由“交易流水号 + 回调事件类型”组成；第二，将去重表和支付入账写入同一个数据库事务，任何一步失败都整体回滚；第三，在监控大盘中加入“去重拦截率”和“重复消息 TOP 来源”两个指标，实时发现新的重复链路。

重复率是衡量幂等效果的关键指标，定义如下：

回到驿站视角：签收台账记得越清晰，越不会在年底盘库时发现多了一堆“不存在”的货。幂等设计不是上线后就一劳永逸，而是要像驿站管理员一样，每天看一遍拦截记录，及时发现异常重试，保证每一件“消息包裹”都只真正入一次库。

AI 应用场景

Python 演示

import sqlite3import time class IdempotentConsumer:    def __init__(self, db_path=":memory:"):        self.conn = sqlite3.connect(db_path)        self.conn.execute("CREATE TABLE IF NOT EXISTS dedup(msg_id TEXT PRIMARY KEY, created_at TEXT)")        self.conn.execute("CREATE TABLE IF NOT EXISTS orders(order_id TEXT PRIMARY KEY, payload TEXT)")        self.conn.commit()     def consume(self, msg):        try:            self.conn.execute("BEGIN IMMEDIATE")            cur = self.conn.execute(                "INSERT INTO dedup(msg_id, created_at) VALUES(?, ?)",                (msg["msg_id"], msg["created_at"]),            )        except sqlite3.IntegrityError:            self.conn.rollback()            return False        if cur.rowcount != 1:            self.conn.rollback()            return False        self.conn.execute(            "INSERT INTO orders(order_id, payload) VALUES(?, ?)",            (msg["order_id"], msg["payload"]),        )        self.conn.commit()        return True now = time.strftime("%Y-%m-%d %H:%M:%S")messages = [    {"msg_id": "20260612-1001-CREATE-01", "order_id": "1001", "payload": "create_order", "created_at": now},    {"msg_id": "20260612-1001-CREATE-01", "order_id": "1001", "payload": "create_order", "created_at": now},    {"msg_id": "20260612-1002-DEDUCT-01", "order_id": "1002", "payload": "deduct_stock", "created_at": now},] consumer = IdempotentConsumer()for msg in messages:    ok = consumer.consume(msg)    print(f"{msg['msg_id']} -> {'处理成功' if ok else '重复消息，已拦截'}")

1. CREATE TABLE IF NOT EXISTS dedup(msg_id TEXT PRIMARY KEY用 msg_id 主键作为去重表的唯一约束。
2. self.conn.execute("BEGIN IMMEDIATE")立即开启写事务，保证去重和业务在同一个原子单元内。
3. INSERT INTO dedup(msg_id, created_at) VALUES(?, ?)先把消息 ID 写入去重表，判断是否第一次出现。
4. except sqlite3.IntegrityError:主键冲突说明之前已处理过，直接回滚并返回 False。
5. self.conn.execute("INSERT INTO orders(...)")去重成功后才执行业务写入，双方一起提交。

常见误区

面试题