deerflow-code/offline-backend-20260512/backend/docs/maf-sse-multi-worker-deployment.md

MAF 会商 SSE 多 Worker 部署说明

状态：Phase 2 (R2) — 单 Worker 已就绪，多 Worker 需补 Redis Pub/Sub 更新时间：2026-08-03

---

1. 当前架构（单 Worker）

ConversationEventHub（app/maf_roundtable/conversation_event_hub.py）是一个 进程内 的 asyncio.Event 通知器：

• 每个 worker 进程有自己独立的 hub 实例（app.state.maf_conversation_event_hub）
• 事件写入者（ConversationService.send_message）在事务提交后调用 hub.publish()
• SSE 订阅者通过 hub.subscribe() 注册一个 asyncio.Event，被 publish() 唤醒后 从数据库查询最新事件（通知-only，不缓存事件内容）
• 心跳：15 秒超时 → : heartbeat\n\n SSE 注释帧

适用场景： uvicorn --workers 1 / gunicorn -w 1，开发环境、小规模灰度。

---

2. 多 Worker 限制

当使用 --workers 2 或更多 worker 时：

问题	原因	影响
跨 Worker 不通知	Worker A 写入事件后 hub.publish() 只通知 A 自己的订阅者；Worker B 的 SSE 客户端收不到实时推送	SSE 流"卡住"直到客户端超时重连，重连后从 DB 重放才能补到事件
无跨进程共享状态	asyncio.Event 是进程内对象，不能跨进程通信	需要 Redis Pub/Sub 或等价中间件

在接入 Redis Pub/Sub 之前，禁止多 Worker 灰度上线。

---

3. 健康检查建议

建议在 GET /api/maf-roundtable/health 中增加：

{
  "sse_hub_type": "in_memory",
  "sse_multi_worker": false,
  "sse_warning": "Single-worker only. Multi-worker requires Redis Pub/Sub."
}

当前版本已通过 sse_multi_worker: false 明确标注不支持多 Worker。 部署脚本应检查此项，在多 Worker 模式下发出告警。

---

4. Redis Pub/Sub 接入方案（后续阶段）

当需要多 Worker 时，替换 ConversationEventHub 为 Redis-backed 实现：

4.1 接口不变

class ConversationEventHub(abc.ABC):
    async def publish(self, conversation_id: str) -> None: ...
    async def subscribe(self, conversation_id: str, ...) -> AsyncIterator[bool]: ...

4.2 Redis 实现

class RedisConversationEventHub(ConversationEventHub):
    def __init__(self, redis: Redis, *, heartbeat_interval: float = 15.0):
        self._redis = redis
        self._heartbeat_interval = heartbeat_interval

    async def publish(self, conversation_id: str) -> None:
        await self._redis.publish(f"maf:conv:{conversation_id}", "1")

    async def subscribe(self, conversation_id: str, ...):
        pubsub = self._redis.pubsub()
        await pubsub.subscribe(f"maf:conv:{conversation_id}")
        try:
            while True:
                msg = await asyncio.wait_for(
                    pubsub.get_message(ignore_subscribe_messages=True, timeout=...),
                    timeout=self._heartbeat_interval,
                )
                if msg is None:
                    yield False  # heartbeat
                else:
                    yield True  # new events might be available
        finally:
            await pubsub.unsubscribe(f"maf:conv:{conversation_id}")

4.3 选择逻辑

# deps.py
if config.database and config.database.redis_url:
    hub = RedisConversationEventHub(create_redis(config.database.redis_url))
else:
    hub = ConversationEventHub()  # in-memory, single-worker

---

5. 部署检查清单

• 确认 uvicorn --workers 1（或等价单进程配置）
• 确认 SSE 端点 GET /api/maf-roundtable/conversations/{id}/stream 可访问
• 确认 nginx / 反向代理未缓冲 SSE（X-Accel-Buffering: no 生效）
• 确认心跳间隔 15s 足够长不会误触发代理超时（proxy_read_timeout 应 ≥ 60s）
• 确认 after_seq 重连不产生重复气泡（前端 reducer 按 seq 去重）
• 多 Worker 上线前：接入 Redis Pub/Sub 并更新 sse_multi_worker: true