单体 NestJS 用 Redis Stream 当事件总线

Nest / 基础设施 · 第 1 篇
玩法地图级说明见自文档《事件驱动与模块解耦》；本文偏 为何用 Stream、消息长什么样、怎么消费不丢不重。

写在前面

自文档 11 已经讲清：Stream 管可靠异步，WebSocket 管在线即时——评论入库、铃铛推送、RPG 加经验是三条节奏。
读者若再问「那为什么不上 RabbitMQ / Kafka？」11 没有展开。本文用现网 blog:events 回答选型与落地，不复述「评论三连发生了什么」那种体验故事。

代码主路径：core/events/ + rpg/listeners/rpg-event.consumer.ts + rag/listeners/rag-event.consumer.ts。

一、为什么是 Redis Stream，而不是先上 MQ

个人博客单体的约束：

对比常见选项：

结论：在「单体 + 已有 Redis + 可预期拆服务」阶段，Stream 是事件总线的甜点区。以后若 RPG 进程独立部署，契约可以不变——只是另一个进程 XREADGROUP 同一条流（Go 侧已有对等 event.Publisher / Consumer）。

不适合塞进 Stream 的：充值到账关弹窗——pay 履约后直接 WS rechargeComplete（11 已写），延迟与「必须在线」属性都不匹配堆积模型。

二、契约：一条流，两个字段

Stream key: blog:events
Entry fields:
  type    → 事件名字符串
  payload → JSON 字符串

发布（失败只打日志，不阻断 HTTP 主路径）：

// BlogEventPublisher
await this.redisService.xAdd(
  BLOG_EVENT_STREAM,
  { type, payload: JSON.stringify(payload) },
  { maxLen: 10000, approximate: true }, // 近似裁剪，防无限涨
);

事件名集中在 BlogEvents（示例）：

Payload 按类型带 uid、articleId、authorUid 等，不传整篇正文——消费者按需查库，避免 Stream 变成第二数据库。

三、双消费者组：同一条流，各干各的

关键点：一条业务事件可以被 多个独立消费者组 各处理一遍。

发文一次 XADD，RPG 与 RAG 互不抢消息：各组有自己的 last-delivered / pending。
这比「一个巨型 consumer 里 if-else 调所有模块」更利于以后把 rag-indexer 拆到独立进程。

消费者名：{hostname}-{pid}，同组多实例时由 Redis 分派未读条目。

循环要点（RPG）：

• XREADGROUP：COUNT 10，BLOCK 5000
• 空闲时：XAUTOCLAIM（需 Redis ≥ 6.2）回收卡住的 pending（空闲 > 60s）
• 专用 ioredis 连接：createDedicatedClient()，避免与短命令抢同一连接上的阻塞读

四、幂等：ACK 之前先「认领处理权」

Handler 失败时 不 ACK，消息留在 pending，可被 claim 重试——所以 必须幂等。

现实现：tryMarkProcessed(streamId, TTL=7天)（Redis SET NX + 前缀 blog:event:done:）。

parse → SET NX done:{id}
  ├─ 非首次 → ACK 并跳过（重复投递）
  └─ 首次 → dispatch → 成功则 ACK
              失败则不 ACK（也不清 done？需注意）

注意取舍：若「已 SET NX 成功」但 dispatch 抛错，done 键已占用，重试可能被当成重复而跳过——这是 至少一次投递 + 业务侧尽量可重入 与 严格 exactly-once 之间的折中。经验加减、索引重建应设计成可安全重跑（或依赖业务键去重，如浏览日 key）。

未知 type / 坏 JSON：直接 ACK，避免毒消息堵死组。

五、和「拆微服务」的过渡关系

当前仍是 Nest 单进程内多个 @Injectable consumer。迁移路径可以是：

1. 契约冻结：type + payload schema 版本化（哪怕先靠 TypeScript 接口）
2. 按组拆进程：先把 rag-indexer 挪到旁路服务，仍读 blog:events
3. 双写期：HTTP 仍发 Stream；新服务只消费，旧 consumer 用 feature flag 关掉对应 case
4. 量级真上来再评估 Kafka——而不是第一天就上

Go 单体已按同一 Stream 名与字段实现对等 Publisher/Consumer，说明契约足够薄，跨语言复用成本低。

踩坑与取舍

1. SQL 导入文章不发事件
   self-doc / 专栏 run-all-*.sql 直写库 → 无 RPG 经验、RAG 需手动重建索引。属预期。

2. Redis < 6.2
   无 XAUTOCLAIM，进程挂掉后的 pending 要人工或升级 Redis。启动时会打 warn。

3. 发布失败吞掉
   主业务 200 了但经验没加：查日志 Failed to publish blog event。监控应盯 publish 失败率与 pending 长度。

4. Stream 与 WS 共用 Redis、键空间不同
   别把 Adapter 的 key 和 blog:events 混在运维脚本里 FLUSHDB。

5. maxLen≈10000
   近似裁剪可能丢掉未被所有组读完的旧消息；组必须跟上流量，或调大 maxLen / 换更持久方案。

小结

下一篇基础设施向可写 Nest 领域模块拆分（计划 N1）；事件体验向继续看自文档 11。

关键路径

• blog-server/src/modules/core/events/blog.events.ts
• blog-server/src/modules/core/events/blog-event.publisher.ts
• blog-server/src/modules/rpg/listeners/rpg-event.consumer.ts
• blog-server/src/modules/rag/listeners/rag-event.consumer.ts
• blog-server/src/modules/core/redis/redis.service.ts（xAdd / xReadGroup / xAutoClaim）