第九章：接入 Redis 消息队列（异步架构设计）

这一章开始，系统从“能处理消息”升级为：

能承载并发、可扩展、不会被AI阻塞的企业级架构

9.1 为什么必须引入异步队列

在上一章的架构中：

企业微信 → FastAPI → Dispatcher → AI → 返回

问题非常明显：

❌ 同步架构的致命问题

1️⃣ AI调用阻塞

用户发消息
   ↓
等待 DeepSeek
   ↓
接口卡住

2️⃣ 企业微信超时限制

企业微信要求：

5秒内必须返回 success

否则：

❌ 重试
❌ 重复消息
❌ 回调失败

3️⃣ 无法抗并发

10个用户 → OK
1000个用户 → 崩

✔ 正确架构：异步解耦

引入 Redis Queue：

企业微信
   ↓
FastAPI（只接收）
   ↓
Redis Queue（缓存消息）
   ↓
Worker（异步处理）
   ↓
AI（DeepSeek）
   ↓
发送回复

9.2 新架构设计

📌 完整链路

企业微信
   ↓
Webhook (FastAPI)
   ↓
Dispatcher
   ↓
Redis LPUSH（入队）
   ↓
立即返回 success
   ↓
Worker BRPOP（消费）
   ↓
LLM处理
   ↓
发送企业微信消息

9.3 Redis设计模型

我们采用最简单可靠模型：

List 队列模型

📌 基本操作

入队（Producer）

LPUSH queue:wecom message

出队（Consumer）

BRPOP queue:wecom

📌 为什么不用 Kafka / RabbitMQ？

因为：

❌ 太重
❌ 运维复杂
❌ 当前阶段不需要

Redis已经足够：

✔ 快
✔ 稳
✔ 简单
✔ 可扩展

9.4 项目结构升级

新增模块：

app/
├── queue/
│   ├── redis_client.py
│   ├── producer.py
│   └── worker.py

9.5 Redis客户端封装

📌 redis_client.py

import redis
import os


redis_client = redis.Redis(
    host=os.getenv("REDIS_HOST", "localhost"),
    port=int(os.getenv("REDIS_PORT", 6379)),
    decode_responses=True
)

9.6 Producer（入队模块）

📌 producer.py

import json
from app.queue.redis_client import redis_client


QUEUE_KEY = "queue:wecom"


def push_message(message: dict):
    """
    写入队列
    """

    redis_client.lpush(
        QUEUE_KEY,
        json.dumps(message, ensure_ascii=False)
    )

9.7 Dispatcher升级（关键改造）

📌 dispatcher.py（接入队列）

from app.queue.producer import push_message


class Dispatcher:

    def dispatch(self, message: dict):

        msg_type = message.get("MsgType")

        # 这里只处理文本
        if msg_type == "text":

            # 直接入队，不处理AI
            push_message(message)

            return {
                "type": "text",
                "reply": "收到消息，正在处理..."
            }

        return {
            "type": "text",
            "reply": "暂不支持该类型"
        }

📌 关键变化

❌ 之前

Dispatcher → AI（同步）

✔ 现在

Dispatcher → Redis Queue（异步）

9.8 Worker设计（核心消费器）

📌 worker.py

import json
import time
from app.queue.redis_client import redis_client
from app.core.llm import ask_llm
from app.services.wecom_service import send_message

QUEUE_KEY = "queue:wecom"


def run_worker():

    print("Worker started...")

    while True:

        _, data = redis_client.brpop(QUEUE_KEY)

        message = json.loads(data)

        print("处理消息:", message)

        user = message.get("FromUserName")
        content = message.get("Content")

        # 调用AI
        reply = ask_llm(content)

        # 发送回复
        send_message(user, reply)

        time.sleep(0.1)

9.9 Worker运行方式

📌 本地运行

python -m app.queue.worker

📌 Docker运行（后续）

worker service 独立容器

9.10 当前系统架构（升级版）

企业微信
   ↓
FastAPI（Webhook）
   ↓
Dispatcher
   ↓
Redis Queue
   ↓
Worker
   ↓
LLM（DeepSeek）
   ↓
WeCom Send API

9.11 架构收益

✔ 性能提升

请求响应时间：毫秒级

✔ 稳定性提升

AI崩溃 ≠ Webhook崩溃

✔ 可扩展性

可以横向扩展 Worker

✔ 解耦成功

Webhook ≠ AI系统

9.12 当前系统状态总结

✅ 已完成

✔ 企业微信接入
✔ 消息解密
✔ Dispatcher分发
✔ Redis队列写入
✔ Worker消费模型

🚧 当前状态

系统已经变成：

👉 事件驱动架构雏形

❌ 还未完成

❌ DeepSeek正式接入
❌ 企业微信发送API
❌ AccessToken管理
❌ Prompt工程

9.13 本章核心总结

这一章完成后，系统已经从：

👉 “同步Web服务”

升级为：

👉 “异步消息驱动系统”

🚀 下一章预告（第十章）

第十章：Worker接入DeepSeek（AI核心引擎）

下一章我们将完成：

Worker
   ↓
DeepSeek API
   ↓
Prompt工程
   ↓
返回结构化回复

👉 到这一章结束：

系统将第一次具备：

“真正的AI对话能力”