[!info] related notes

• 课程 MOC：BodySense Redis 从零到一课程 MOC
• 前置课程：第 8 课
• 对应知识笔记：Redis Pub/Sub、消息投递语义
• BodySense：BodySense Consultation Streaming Architecture

BodySense Redis 第 9 课：Pub/Sub 与实时通知

[!abstract] 学习目标 理解 Redis Pub/Sub 最适合做“现在有变化，快醒来看看”的通知层，而不是可靠事件日志。

先从零解释：Pub/Sub 是什么

Pub/Sub = Publish / Subscribe，中文通常叫“发布/订阅”。

它描述的是一种消息通信模型：

Publisher
→ 把消息发布到某个 Channel

Subscriber
→ 订阅这个 Channel
→ 当新消息发布时立即收到

三组词先分清。

Publisher

发送消息的一方。

它通常不需要知道当前具体有哪些 Subscriber：

publish("run:123", "changed")

Channel

消息主题/频道名称，例如：

run:123
user:42:notifications
jobs:ready

Channel 不是 Redis Key-Value 数据结构；它更像实时广播路径。

Subscriber

订阅某个 Channel 并保持连接等待消息的一方。

SUBSCRIBE run:123
↓
connection stays subscribed
↓
future messages arrive

Pub/Sub 不是 SSE，也不是 WebSocket

这三个概念很容易混。

Redis Pub/Sub

backend service
↔ Redis
↔ backend service

它主要解决服务进程之间的实时通知/fan-out。

SSE

Server
→ HTTP response stream
→ Browser

它是服务端到浏览器的传输方式。

WebSocket

Browser
↔ persistent duplex connection
↔ Server

它是双向网络传输协议。

所以 BodySense 可以同时：

Redis Pub/Sub
→ 唤醒 Go API instance

Go API
→ SSE
→ Browser

这些不是互相替代关系。

Delivery Semantics：消息系统要问的第一件事

消息系统不能只问：

能不能把消息发出去？

还要问：

订阅者离线怎么办？
消息会保存多久？
失败会不会重试？
会不会重复？
有没有 ACK？
能不能 replay？

这些问题统称 delivery semantics。

Redis 普通 Pub/Sub 的核心心智模型是：

实时在线广播，而不是 durable message log。

BodySense 当前痛点

Consultation 的 active run reattach 当前需要从 PostgreSQL runtime-event log 恢复。

一种简单实现是：

SSE connection
↓
每 250ms
↓
SELECT runtime_events WHERE seq > lastSeq

优点：

• 正确；
• durable；
• reconnect 可恢复。

问题：

即使没有新事件，空闲连接也会反复查询数据库。

连接数增长后，数据库压力与真实 event rate 脱钩。

Redis Pub/Sub 的最小命令模型

Publisher：

PUBLISH run:123 changed

可以读成：

在 channel = run:123
发布 payload = changed

Subscriber：

SUBSCRIBE run:123

只要 Subscriber 当前在线并保持订阅连接：

Publisher
→ Redis
→ fan-out
   ├─ Subscriber A
   ├─ Subscriber B
   └─ Subscriber C

这就是 fan-out：一条发布可以同时分发给多个当前订阅者。

Channel 不会像 Stream 一样保存历史 Entry

你不能把 Pub/Sub 想成：

channel
├─ message 1
├─ message 2
└─ message 3

然后晚来的 subscriber 再从 message 1 开始读。

普通 Pub/Sub 更像广播：

现在开着收音机
→ 听到现在的广播

刚才没开
→ 刚才那段不会自动重播

At-most-once：为什么“没收到”不会自动重发

At-most-once 可以先按字面理解：

对某个 Subscriber 来说，一条消息最多交付一次；如果交付时不在线/连接中断，不保证之后补发。

例如：

12:00:00 subscriber online
12:00:01 receives message A

12:00:02 subscriber disconnects
12:00:03 publisher sends message B

12:00:05 subscriber reconnects

普通 Pub/Sub 不会因为它重新上线，就自动补 message B。

所以 Pub/Sub 不提供：

durable history
consumer offset
pending list
ACK
replay cursor

这也是为什么下一课会引出 Redis Streams。

At-most-once 不等于网络世界绝对不会出现重复业务动作

架构仍应让通知处理 idempotent：

收到 wakeup
→ 去 durable source 查 afterSeq

这样即使应用层因 reconnect/重订阅等原因发生重复 wakeup，也不会重复创建 durable event。

Notification Plane vs Durable Log

推荐架构：

Runtime
↓
PostgreSQL COMMIT RuntimeEvent(seq=N)
↓
Redis PUBLISH run:<runID> "changed"
↓
Go SSE subscriber wakes
↓
SELECT runtime_events WHERE seq > lastSeq
↓
send durable ordered events to browser

Redis 消息只表达：

“有东西变了，你现在可以去查 durable source。”

真正的数据仍是 PostgreSQL：

runtime_events
+ seq
+ afterSeq replay

为什么顺序必须是 Commit Then Notify

错误：

PUBLISH
↓
subscriber wakes
↓
Postgres transaction 还没 commit / 最后失败

会产生虚假唤醒和竞态。

更合理：

PostgreSQL COMMIT success
↓
best-effort PUBLISH

如果：

DB success
Redis publish fails

事件仍然存在。

之后 fallback reconciliation / polling 仍能发现。

如果：

DB fails

就不应该产生“真实事件已经存在”的假象。

Pub/Sub 丢通知为什么不等于丢事件

假设 seq：

lastSeq = 10
Postgres committed seq = 11
Pub/Sub wakeup for 11 lost

只要系统还有：

• fallback timer；
• reconnect；
• 下一次 wakeup；
• afterSeq reconciliation；

就会执行：

SELECT ...
WHERE seq > 10

重新发现 11。

所以：

通知可以丢，事实不能丢。

这是一个非常强的架构模式：

durable truth
+
lossy wakeup optimization

Duplicate Notification 也应安全

Pub/Sub 可能连续通知两次：

changed
changed

Subscriber 不应该：

把消息 payload 当 event 直接 append 两次

而应该：

wakeup
→ query afterSeq

如果没有新的 durable event：

query empty
→ no-op

因此 notification 可以被 coalesce / duplicate，而业务仍正确。

为什么 Pub/Sub 对多 API 实例特别有价值

假设 BodySense 水平扩容：

API A
API B
API C

某个用户的 SSE connection 可能挂在 API B 上，但 Runtime event producer 可能运行在另一个进程。

进程内 channel：

producer in A
→ only A memory
→ B 不知道

Redis Pub/Sub：

producer
↓ PUBLISH run:123
Redis
├─ API A subscribed
├─ API B subscribed
└─ API C subscribed

于是通知跨实例传播。

但这里仍要控制订阅粒度：

所有 instance 订阅所有 run？
每个 active run 单独 channel？
用 pattern subscription？
连接/订阅生命周期如何清理？

这些属于实现设计，而不是 Pub/Sub 自动替你决定。

BodySense 这里的状态：候选实时优化，不是 durable authority

这条课程设计的推荐方向是：

PostgreSQL runtime_events
= 已有 durable replay truth

Redis Pub/Sub
= future/optional wakeup optimization

不要因为学会 PUBLISH/SUBSCRIBE，就把现有 runtime_events + seq 删除。

Prediction

[!question]

1. Subscriber 离线期间的 Pub/Sub 消息会自动保存吗？
2. 为什么 RuntimeEvent payload 不应该只放在 Pub/Sub 中？
3. DB commit 成功但 PUBLISH 失败，BodySense 是否应该丢事件？
4. PUBLISH 成功但 subscriber 晚上线，如何恢复？
5. 为什么 Pub/Sub message duplicated 也不应造成重复业务事件？

[!success]- Answer / 解析 1. Subscriber 离线期间的 Pub/Sub 消息会自动保存吗？
不会。普通 Redis Pub/Sub 面向当前在线订阅者，是瞬时广播；订阅者离线期间发布的消息不会等它回来再补发。

2. 为什么 RuntimeEvent payload 不应该只放在 Pub/Sub 中？
因为 Pub/Sub 没有 durable history、ACK、Pending 和离线 replay。只把事件放在 Pub/Sub 中，一次网络抖动、进程重启或晚订阅就可能造成永久业务事件丢失。BodySense 的 durable payload 应继续由 PostgreSQL runtime_events + seq 持有。

3. DB commit 成功但 PUBLISH 失败，是否应该丢事件？
不应该。正确架构是“durable commit → best-effort notify”。事件已经存在 PostgreSQL，Redis 只损失即时唤醒；之后可以依赖 reconciliation polling、下一条通知或 reconnect + after_seq 找回。

4. PUBLISH 成功但 Subscriber 晚上线，如何恢复？
不尝试从 Pub/Sub 补历史，而是拿自己的 lastSeq 去 durable event log 查询 seq > lastSeq。也就是说恢复依赖 PostgreSQL cursor，而不是 Pub/Sub retention。

5. 为什么 Pub/Sub notification 重复也不应产生重复业务事件？
因为 notification 只表达“有变化，去查一下”。Subscriber 醒来后仍从 durable source 按 afterSeq 查询；如果没有新 seq 就 no-op。如果有新事件，也由 seq / event identity 保证同一事实不会因为通知重复而重复创建。

Failure Injection

验证：

publish success
publish lost
subscriber starts late
Redis unavailable
Redis reconnect
multiple API replicas
duplicate notification
multiple events before one wakeup
terminal stream.done

每次都验证：

Postgres replay remains complete
+
seq ordered
+
no duplicate business effect

L4 Gate

能解释：

为什么 Redis Pub/Sub 可以显著减少空轮询，但绝不能替代 runtime_events 表。

并画出：

durable write
→ best-effort notify
→ wake
→ durable reconcile