第十二章：企业微信消息完整闭环优化（可靠性与重试机制）
这一章开始，系统从：
“能跑的 AI 系统”
升级为：
“生产级可靠消息系统”
12.1 当前系统的真实问题
虽然我们已经完成：
企业微信 → FastAPI → Redis → Worker → AI → 回复
但在生产环境中，还会遇到这些问题：
❌ 1. 消息丢失
Worker宕机 → Redis数据未处理
❌ 2. AI调用失败
DeepSeek超时 / 网络波动
❌ 3. 重复消息
企业微信重试机制 → 同一消息多次进入
❌ 4. Worker崩溃
异常未捕获 → 消费中断
12.2 本章目标
构建一个完整的：
✔ 幂等机制
✔ 重试机制
✔ 死信队列
✔ 异常恢复
✔ 消息去重
12.3 核心设计思想
📌 企业级消息处理标准
至少一次（At least once）
+ 幂等处理（Idempotency）
+ 可重试（Retryable）
+ 可追踪（Traceable）
12.4 幂等机制设计（MsgId）
企业微信每条消息都有：
MsgId（唯一）
📌 Redis去重设计
Key:
wecom:msg:{MsgId}
Value:
1（已处理）
📌 去重逻辑
def is_duplicate(msg_id: str):
key = f"wecom:msg:{msg_id}"
if redis_client.get(key):
return True
redis_client.set(key, 1, ex=86400) # 24小时
return False
12.5 Worker升级（幂等控制）
📌 worker.py（增强版）
import json
import asyncio
from app.queue.redis_client import redis_client
from app.core.llm import LLMClient
from app.services.wecom_service import send_message
QUEUE_KEY = "queue:wecom"
llm = LLMClient()
def is_duplicate(msg_id: str):
key = f"wecom:msg:{msg_id}"
if redis_client.get(key):
return True
redis_client.set(key, 1, ex=86400)
return False
async def process(message: dict):
msg_id = message.get("MsgId")
# 🚨 幂等检查
if msg_id and is_duplicate(msg_id):
print("重复消息，跳过:", msg_id)
return
user = message.get("FromUserName")
content = message.get("Content")
try:
print("处理消息:", content)
reply = await llm.chat(content)
await send_message(user, reply)
except Exception as e:
print("处理失败:", e)
# ❗进入重试队列
redis_client.lpush("queue:wecom_retry", json.dumps(message))
def run_worker():
print("Worker started...")
while True:
_, data = redis_client.brpop(QUEUE_KEY)
message = json.loads(data)
asyncio.run(process(message))
12.6 重试队列设计
📌 新队列
queue:wecom_retry
📌 retry worker（独立逻辑）
def retry_worker():
print("Retry Worker started...")
while True:
_, data = redis_client.brpop("queue:wecom_retry")
message = json.loads(data)
retry_count = message.get("retry", 0)
if retry_count > 3:
redis_client.lpush("queue:wecom_dead", json.dumps(message))
continue
message["retry"] = retry_count + 1
redis_client.lpush("queue:wecom", json.dumps(message))
12.7 死信队列（DLQ）
📌 dead letter queue
queue:wecom_dead
用途：
- ❌ 永久失败消息
- ❌ 人工介入
- ❌ 日志分析
📌 DLQ数据结构
{
"MsgId": "xxx",
"Content": "xxx",
"error": "timeout",
"retry": 3
}
12.8 消息完整生命周期
企业微信
↓
FastAPI
↓
Redis Queue
↓
Worker
↓
AI处理
↓
成功 → 回复
↓
失败 → Retry Queue
↓
超过次数 → Dead Queue
12.9 系统可靠性提升
✔ 改造前
❌ 失败 = 丢失
❌ 无重试
❌ 无记录
✔ 改造后
✔ 自动重试
✔ 自动恢复
✔ 去重处理
✔ 死信记录
12.10 幂等 + 重试组合价值
📌 企业级标准能力
✔ Exactly-once（伪实现）
✔ At-least-once（真实保证）
✔ Fault tolerance（容错）
12.11 当前系统架构（最终稳定版）
企业微信
↓
FastAPI Webhook
↓
Redis Queue
↓
Worker
↓
AI（DeepSeek）
↓
WeCom Send API
↓
Token Manager
↓
Redis缓存
失败路径：
Worker失败 → Retry Queue → Dead Queue
12.12 当前系统能力总结
✅ 已完成能力
✔ AI对话系统
✔ 异步队列架构
✔ Worker模型
✔ DeepSeek接入
✔ Token缓存
✔ 幂等机制
✔ 重试机制
✔ 死信队列
✔ 系统容错
12.13 本章核心价值
这一章完成后：
👉 系统从“能用”
升级为
👉 “企业级可靠消息系统”
🚀 下一章预告（第十三章）
第十三章：系统日志、TraceID与可观测性设计
下一章我们将解决生产环境最关键的问题：
❌ 无法定位问题
❌ 无法追踪消息
❌ Worker黑盒
❌ AI调用不可观测
下一章核心内容：
✔ 全链路TraceID
✔ 结构化日志
✔ 请求追踪
✔ Worker日志系统
✔ AI调用记录
👉 到这一章结束：
系统将具备：
“可观测 + 可调试 + 可追踪能力”