消息队列的幂等设计重复消费不是 bug是分布式系统的常态一、为什么一条消息会被消费两次使用消息队列时最常见的认知偏差是以为publish一次就对应consume一次。但在分布式系统中消息的 at-most-once、at-least-once 和 exactly-once 三种投递语义中只有 exactly-once 是理想目标实际落地往往是 at-least-once。原因在于网络超时导致生产者重发消费者处理成功但确认失败ack 超时导致消息被重新投递Broker 主从切换时的消息重复这意味着重复消费不是异常而是常态。你的消费逻辑必须幂等——一条消息无论被消费多少次最终的业务效果与消费一次相同。二、幂等设计的思路框架flowchart TD A[收到消息] -- B[提取消息唯一 ID] B -- C{去重表中查询} C --|已处理| D[直接返回成功 ack] C --|未处理| E[执行业务逻辑] E -- F[记录处理结果到去重表] F -- G[返回 ack] E -- H{业务失败?} H --|是| I[记录失败状态] I -- J[NACK 或 进入死信队列]核心思想在消息 ID 级别做去重。每条消息需要一个全局唯一的业务 ID用这个 ID 判断消息是否已被处理过。三、实现幂等消费者框架import hashlib import json import time from dataclasses import dataclass from abc import ABC, abstractmethod from typing import Optional from redis import Redis dataclass class Message: 消息结构 message_id: str # 全局唯一消息 ID business_key: str # 业务幂等键如 order_id payload: dict # 消息体 timestamp: float # 消息时间戳 class IdempotentConsumer(ABC): 幂等消费者基类 所有需要幂等消费的业务逻辑都应继承此基类。 子类只需实现 _process 方法幂等逻辑由基类处理。 设计思路 1. 用 Redis 做去重存储性能 TTL 自动清理 2. 去重 key 格式idempotent:{服务名}:{business_key} 3. 记录处理状态PENDING → SUCCESS / FAILED # 去重记录保留时间秒 # 设置合理值既要覆盖消息重试窗口又要及时释放内存 DEDUP_TTL 7 * 24 * 3600 # 7 天 def __init__(self, redis_client: Redis, service_name: str): self.redis redis_client self.service_name service_name def consume(self, message: Message) - bool: 消费消息幂等入口 Returns: True: 消息处理成功可 ack False: 消息已处理过幂等直接 ack dedup_key self._dedup_key(message.business_key) # 第一步检查是否已处理 previous_status self._check_previous(message.business_key) if previous_status SUCCESS: # 已成功处理过直接返回 return True if previous_status IN_PROGRESS: # 正在处理中可能是另一个消费者在处理 # 等待一段时间后重试 time.sleep(0.5) return self.consume(message) # 第二步标记处理中防止并发重复处理 # 使用 SETNX 实现分布式锁 acquired self.redis.set( dedup_key, IN_PROGRESS, nxTrue, # 仅当 key 不存在时设置 ex60, # 60 秒超时防止死锁 ) if not acquired: # 其他消费者已获取了处理权 return True try: # 第三步执行实际业务逻辑 result self._process(message) # 第四步记录成功 self.redis.setex( dedup_key, self.DEDUP_TTL, SUCCESS, ) return True except Exception as e: # 第五步记录失败可选根据业务需求 self.redis.setex( f{dedup_key}:error, self.DEDUP_TTL, json.dumps({error: str(e), time: time.time()}), ) # 删除 IN_PROGRESS 标记允许重试 self.redis.delete(dedup_key) raise # 重新抛出让 MQ 框架决定重试策略 abstractmethod def _process(self, message: Message) - any: 子类实现实际的业务逻辑 pass def _dedup_key(self, business_key: str) - str: 生成去重 key return fidempotent:{self.service_name}:{business_key} def _check_previous(self, business_key: str) - Optional[str]: 检查之前的处理状态 status self.redis.get(self._dedup_key(business_key)) return status.decode() if status else None # ---- 使用示例积分变更消息的幂等消费 ---- class PointsChangeConsumer(IdempotentConsumer): 积分变更消费者保证同一笔积分变更不会被重复处理 def __init__(self, redis_client, db_client): super().__init__(redis_client, points_change) self.db db_client def _process(self, message: Message) - None: 处理积分变更 通过 business_keyorder_id保证幂等 同一个订单的积分变更只会被执行一次。 order_id message.payload[order_id] user_id message.payload[user_id] points message.payload[points] # 使用数据库事务 乐观锁 with self.db.transaction(): # SELECT ... FOR UPDATE 锁定行 existing self.db.query( SELECT id FROM points_log WHERE order_id ?, [order_id], ) if existing: # 已处理过跳过 return # 更新积分 self.db.execute( UPDATE users SET points points ? WHERE id ?, [points, user_id], ) # 记录日志 self.db.execute( INSERT INTO points_log(order_id, user_id, points) VALUES(?, ?, ?), [order_id, user_id, points], )四、幂等 key 的设计策略4.1 天然幂等键有些业务天然有唯一 ID直接使用即可订单号order_id支付流水号transaction_id请求 IDrequest_id由上游生成4.2 计算幂等键没有天然 ID 时对消息关键字段做哈希def compute_idempotent_key(message: dict) - str: 从消息内容计算幂等键 # 选取影响唯一性的字段 key_fields [ message.get(user_id), message.get(action), message.get(timestamp), ] raw |.join(str(f) for f in key_fields) return hashlib.md5(raw.encode()).hexdigest()注意时间戳作为幂等键的一部分会导致重发消息产生不同的幂等键。如果需要重试应使用业务时间如订单创建时间而非消息时间。4.3 幂等键的粒度粒度太粗如user_id不同操作会误判为重复。粒度太细如所有字段的哈希正常重试也会被当作新消息。正确的粒度是业务操作级别的唯一标识——对于积分扣减是order_id对于用户注册是email或phone。五、边界与权衡5.1 Redis 不可用时如果 Redis 宕机去重会失败。降级策略是允许重复处理的风险同时告警通知快速修复。绝对不允许因为去重服务不可用而导致消费完全停止。5.2 去重 TTL 的设置TTL 太短1 小时消息重试窗口内去重记录已过期无法拦截重复。TTL 太长30 天Redis 内存压力大。推荐设置为消息的最大重试窗口的 3~5 倍。如果消息最大重试时间为 24 小时TTL 设 3~7 天是合理的。5.3 并发幂等当多个消费者同时收到同一条消息的重复副本时SETNX的分布式锁机制保证只有一个消费者会实际执行逻辑。其余消费者在检查到IN_PROGRESS状态后会等待或直接返回。5.4 数据库层面的幂等最彻底的幂等是在数据库层面通过唯一约束来保证。积分日志表的order_id设置为 UNIQUE即使应用层去重失败数据库也会拒绝重复写入。六、总结消息队列的幂等设计是一个必须假设重复会发生的工程实践。核心思路是用一个唯一键业务 ID 或计算出的哈希来标识每次操作在消费前检查、消费后记录。Redis 数据库双保险是最常见的落地方案。理解 at-least-once 的代价是理解分布式系统可靠性的第一步。