企业官网建设流程全解析

消息队列的幂等设计：重复消费不是 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", nx=True, # 仅当 key 不存在时设置 ex=60, # 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 f"idempotent:{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_key（order_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）
• 请求 ID（request_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 的代价，是理解分布式系统可靠性的第一步。