第八章：消息分发与 Dispatcher 设计（系统进入可扩展阶段）
这一章是整个系统从“能解析消息”到“能处理业务逻辑”的分水岭。
上一章结束时，我们已经做到：
企业微信 → 解密 → 明文XML → Python对象
但目前的问题是：
所有消息都堆在 callback 里处理
这在工程上是不可接受的。
8.1 为什么必须要 Dispatcher
如果不做分发，会变成这样：
if text:
do_text()
elif event:
do_event()
elif image:
do_image()
问题：
- ❌ 代码越来越臃肿
- ❌ 无法扩展新消息类型
- ❌ AI逻辑混入入口层
- ❌ 不可测试
- ❌ 不可拆分部署
✔ 正确做法：Dispatcher模式
核心思想：
“入口只负责接收，处理交给不同模块”
8.2 系统分层设计
本章完成后的结构：
callback.py
↓
dispatcher.py
↓
handler/
├── text.py
├── event.py
├── image.py
└── file.py
8.3 Dispatcher设计
📌 dispatcher.py（核心）
from app.handlers.text import handle_text
from app.handlers.event import handle_event
from app.handlers.image import handle_image
class Dispatcher:
def dispatch(self, message: dict):
msg_type = message.get("MsgType")
if msg_type == "text":
return handle_text(message)
elif msg_type == "event":
return handle_event(message)
elif msg_type == "image":
return handle_image(message)
else:
return self.handle_unknown(message)
def handle_unknown(self, message):
print("未知消息类型:", message)
return None
8.4 Handler设计原则
每个 handler 必须满足：
✔ 单一职责
✔ 无外部依赖
✔ 可独立测试
✔ 可独立扩展
8.5 文本消息处理（核心入口）
📌 handlers/text.py
from app.core.llm import ask_llm
def handle_text(message: dict):
"""
文本消息处理
"""
user_id = message.get("FromUserName")
content = message.get("Content")
print(f"[TEXT] user={user_id}, content={content}")
# 这里先不接AI，只做结构验证
return {
"type": "text",
"reply": f"收到：{content}"
}
8.6 事件消息处理
📌 handlers/event.py
def handle_event(message: dict):
"""
事件消息处理
"""
event_type = message.get("Event")
print("[EVENT]", event_type)
if event_type == "subscribe":
return {
"type": "text",
"reply": "欢迎关注企业AI助手"
}
return {
"type": "text",
"reply": "事件已收到"
}
8.7 图片消息处理
📌 handlers/image.py
def handle_image(message: dict):
"""
图片消息处理
"""
pic_url = message.get("PicUrl")
print("[IMAGE]", pic_url)
return {
"type": "text",
"reply": "已收到图片"
}
8.8 callback接入Dispatcher
📌 callback.py（最终整合）
from fastapi import APIRouter, Request
from app.wecom.crypto import WeComCrypto
from app.wecom.parser import parse_encrypt_xml, parse_plain_xml
from app.handlers.dispatcher import Dispatcher
router = APIRouter()
crypto = WeComCrypto()
dispatcher = Dispatcher()
@router.post("/callback")
async def callback(request: Request):
params = dict(request.query_params)
msg_signature = params.get("msg_signature")
timestamp = params.get("timestamp")
nonce = params.get("nonce")
body = await request.body()
xml_str = body.decode("utf-8")
# 1️⃣ 外层解析
outer = parse_encrypt_xml(xml_str)
encrypt = outer.get("Encrypt")
# 2️⃣ 校验 + 解密
crypto.verify_signature(msg_signature, timestamp, nonce, encrypt)
plain_xml = crypto.decrypt_message(
msg_signature,
timestamp,
nonce,
xml_str
)
# 3️⃣ 明文解析
message = parse_plain_xml(plain_xml)
# 4️⃣ 分发
result = dispatcher.dispatch(message)
return "success"
8.9 当前系统能力状态
✅ 已完成能力
✔ 企业微信接入
✔ 消息解密
✔ XML解析
✔ 消息结构化
✔ Dispatcher路由
✔ Handler拆分
🚧 当前系统状态
企业微信消息流：
✔ 已进入系统
✔ 已可识别类型
✔ 已可分模块处理
但：
❌ 还没有AI
❌ 还没有Redis队列
❌ 还没有异步处理
8.10 Dispatcher的工程价值
这一章建立了一个非常关键的能力：
系统从“写死逻辑”变成“可扩展架构”
📌 未来扩展能力
以后你可以直接加：
voice.py
video.py
file.py
location.py
无需修改 callback。
8.11 本章总结
这一章完成后，系统已经具备：
👉 “可扩展消息处理架构”
入口层彻底解耦，业务开始模块化
🚀 下一章预告（第九章）
第九章：接入Redis消息队列（异步架构设计）
下一章我们会解决一个关键问题：
现在系统是同步处理
❌ callback等待AI
❌ 响应慢
❌ 无法扩展
下一步我们将引入：
Redis Queue
Producer / Consumer
Worker异步模型
👉 系统将从：
同步机器人 → 事件驱动架构系统