BodySense Redis 第 10 课：Streams、Queue 与 Worker

从 List Queue 到 Redis Streams、Consumer Group、ACK、Pending 与 retry，结合 BodySense PostgreSQL JobRuntime 判断 Redis 应做 Job authority 还是 wakeup/queue substrate。

#course #growing #backend #architecture #redis #bodysense

[!info] related notes

课程 MOC：BodySense Redis 从零到一课程 MOC

前置课程：第 9 课

对应知识笔记：Redis Streams、消息投递语义、异步任务队列

消息地图：Messaging & Queues MOC

BodySense Redis 第 10 课：Streams、Queue 与 Worker

[!abstract] 学习目标学会区分“通知”“队列”“durable job authority”，并理解为什么一个项目已经有可靠 PostgreSQL JobRuntime 时，不应因为学了 Redis Streams 就立刻迁移全部后台任务。

先从零解释：Queue、Job、Worker 分别是什么

这一课第一次进入“后台任务系统”，先不要把 Queue、Stream、Worker 混成一个词。

Job / Task

一份“以后需要被执行的工作描述”。

例如：

generate diagnosis
process upload
send email
rebuild projection

Job 通常至少需要 identity 和 payload；生产系统还经常需要 status、attempt、deadline、idempotency 等状态。

Producer

创建 Job / Message 的一方。

HTTP request
→ create work
→ enqueue / persist

Queue

用来把“产生工作”和“执行工作”解耦。

Producer
→ Queue
→ Worker

有了 Queue，Producer 不必等任务全部执行完再返回。

Worker / Consumer

从 Queue 或 Log 中取工作并执行的一方。

take job
→ execute
→ success / failure
→ ACK / retry / state transition

所以真正的任务系统不是“能把一个 JSON 放进 Redis”就结束了。

至少还要回答：

任务有没有持久化？
谁现在拥有它？
Worker 崩溃怎么办？
失败重试几次？
执行过但 ACK 丢了怎么办？
重复执行是否安全？
毒任务怎么隔离？
怎么观察积压？

从最简单的 List Queue 开始

Redis List 是一个有顺序的元素序列，可以从两端 push/pop。

一个极简 Queue：

LPUSH jobs job-1
BRPOP jobs 0

可以理解成：

LPUSH
→ Producer 把任务放进去

BRPOP
→ Worker 如果没任务就阻塞等待
→ 有任务时从另一端取走

B 的意思是 Blocking：当 List 为空时，不需要 Worker 每 100ms 主动轮询；连接可以等到有元素。

为什么“取出来”不等于可靠完成

最简单 pop queue 有一个明显问题：

Worker BRPOP job-1
↓ job 已离开 queue
Worker crashes before processing
↓
job-1 去哪里了？

如果没有额外 reservation/backup list/ack protocol，这个任务可能已经从主 Queue 消失。

因此复杂任务系统需要更多状态：

delivery / reservation；
ACK；
pending；
retry；
ownership；
dead-letter；
idempotency。

这就是 Redis Streams 比“一个 List”更值得继续学的原因。

Redis Streams：不是网络视频流，而是保留 Entry 的日志型数据结构

Redis Stream 这个名字容易让人联想到 SSE/HTTP streaming。

这里不是那个意思。

可以先把 Stream 想成：

一个按 ID 排列、可以不断 append 新 Entry、还能回头读取历史 Entry 的 Redis 数据结构。

追加：

XADD jobs * type upload id 123

* 表示让 Redis 自动生成 Entry ID。

结果类似：

1726660000000-0
  type = upload
  id   = 123

Entry ID 是什么

默认 ID 常呈现：

<millisecondsTime>-<sequence>

例如：

1726660000000-0
1726660000000-1
1726660000001-0

它提供了稳定的排序/游标位置。

为什么它和 Pub/Sub 根本不同

Pub/Sub：

message exists only as live delivery

Stream：

entry appended into data structure
→ 可以之后按 ID 再读

所以晚来的消费者可以读取 retained history，而不是只能等“下一条”。

最基础读取 XREAD

概念上：

XREAD STREAMS jobs <last-id>

表达：

从我上次看到的位置之后继续读取。

这已经开始像 durable cursor，而不是广播。

Consumer Group：让多个 Worker 作为一个消费团队分工

如果三个 Worker 都普通 XREAD 同一个 Stream：

worker A → 看历史
worker B → 也看历史
worker C → 也看历史

这更像多个独立订阅者。

Consumer Group 解决的是：

一组 Worker 共同处理一份工作流，让新 Entry 分配给组内不同 Consumer，而不是每个人都处理一遍。

模型：

Stream: jobs
↓
Consumer Group: workers
├─ consumer-a
├─ consumer-b
└─ consumer-c

创建 group：

XGROUP CREATE jobs workers 0

消费：

XREADGROUP GROUP workers consumer-a STREAMS jobs >

这里的 > 大致表达：

给我这个 group 中尚未投递给其他 consumer 的新 Entry。

Group 和 Consumer 是两层 identity

Group：

整个 worker fleet 的消费进度/工作集合

Consumer：

group 内具体哪个 worker instance 领取了 Entry

这样 Redis 才能知道某条消息当前挂在哪个 consumer 名下。

ACK 与 Pending Entries：收到，不等于完成

Worker 通过 XREADGROUP 收到 Entry 后，Redis 会把它视作已经 delivery，但还没有完成。

可以画成：

new entry
↓ XREADGROUP
consumer-a receives
↓
Pending Entries List (PEL)
↓
业务处理成功
↓ XACK
done for this group

ACK 是什么

ACK = acknowledgement，确认。

它表达：

这个 Consumer Group 已经成功处理这条消息，不需要再把它当作 pending work。

命令：

XACK jobs workers <entry-id>

为什么需要 Pending

如果 Worker：

receive m1
→ crash before ACK

Redis 仍知道：

m1 已经投递
但没有 ACK

因此可以：

XPENDING jobs workers

看到 pending 状态。

后续另一个 Worker 可以 claim/recover 超时的 Entry。

Claim 的意义

不是“复制一份新消息”，而是：

这个旧 Consumer 可能死了，把它未完成的 pending ownership 转给健康 Consumer。

这就是任务恢复协议的一部分。

Delivery Semantics：At-most-once、At-least-once、Exactly-once

这一组词经常在消息系统面试里出现。

At-most-once

最多处理一次
但可能完全没处理

典型风险是丢。

At-least-once

保证尽量最终处理
为了恢复可能重复 delivery

典型风险是重复。

Redis Streams + Pending/Claim 很容易进入这种语义：

没 ACK
→ 之后重新处理

Exactly-once

听起来是：

业务副作用只发生一次

但真正端到端 exactly-once 很难，因为消息 ACK 和外部副作用通常不在一个原子事务里。

经典窗口：

Worker receives m1
↓
INSERT PostgreSQL success
↓
process crashes
↓
XACK never happened
↓
m1 redelivered
↓
INSERT again

所以：

消息层的 ACK 机制 ≠ 业务层 exactly-once。

生产中更常见目标是：

at-least-once delivery
+
idempotent business effect

例如使用 unique key / idempotency key，让重复消费得到同一个业务结果。

Pub/Sub vs Stream

	Pub/Sub	Stream
离线后历史	不保留给普通 subscriber	可以读取历史
Consumer Group	无	有
Pending / ACK	无	有
典型用途	在线通知 / fan-out	事件、任务、消费进度
BodySense 候选	SSE wakeup	queue/worker isolated lab

BodySense 当前 JobRuntime

当前 PostgreSQL 已经持有：

jobs
job_events
idempotency_key
attempts
max_attempts
legal transition
ClaimPending
recovery sweep

这意味着它已经不是“没有任务系统，所以需要 Redis”。

问题更像：

周期性 DB polling 是否产生不必要的启动等待？

更低风险的目标架构

Create Job transaction
↓ PostgreSQL COMMIT
Redis notify job-ready
↓
Worker wakes immediately
↓
PostgreSQL ClaimPending

Redis 丢掉通知：

periodic DB recovery sweep
→ eventually claim

这样：

Redis
= latency optimization / wakeup

PostgreSQL
= job authority / retry budget / legal state / idempotency

为什么这比直接迁移风险低

因为现有正确性路径仍保留：

Redis completely down
→ worker latency may increase
→ DB sweep still recovers work

而不是：

Redis down
→ job state itself disappeared / inaccessible

这是上一课“durable truth + lossy wakeup”模式在 Job Runtime 上的延伸。

DB Transaction + Queue 双写问题：为什么“写 DB + 发消息”很难一起保证

假设创建 Job 需要：

1. PostgreSQL INSERT job
2. Redis enqueue job-ready

这其实是两个独立系统。

失败窗口一：

PostgreSQL COMMIT success
↓
Redis enqueue fails

结果：

Job durable exists
但 queue 中没有消息

失败窗口二：

Redis enqueue success
↓
PostgreSQL transaction rolls back

结果：

Worker 收到一个
数据库里不存在/未提交的 Job

Transactional Outbox 是什么

如果业务要求“DB commit 后消息必须可靠最终发出”，常见模式是：

同一个 PostgreSQL transaction
├─ 写业务表
└─ 写 outbox row
↓ COMMIT

独立 relay
→ 扫描/订阅 outbox
→ 发布消息
→ 标记已处理

这样至少把：

business fact
+
message intent

放进一个数据库原子提交。

发布本身仍需处理重复，因此 Consumer 仍应幂等。

为什么 BodySense 当前不一定需要 Outbox 才能做 Job Wakeup

如果 Redis 只承担：

best-effort wakeup

而 PostgreSQL Worker 本来就有：

periodic recovery sweep

那么：

DB COMMIT
→ try Redis notify
→ notify lost
→ later DB sweep still sees Job

正确性仍在。

此时 Outbox 可能是过度设计，除非以后通知本身升级为必须可靠的 integration event。

Asynq 的定位

Asynq 是 Redis 驱动的 Go 分布式任务队列库，可以提供：

enqueue；
retry；
scheduled/delayed task；
worker；
task state。

但引入它意味着：

任务 ownership 和运维依赖发生变化。

所以先在 isolated lab 学，而不是直接替换 JobRuntime。

Lab

Redis Streams：

XADD
XGROUP CREATE
XREADGROUP
XACK
XPENDING
XCLAIM / recovery

观察：

正常消费；
Consumer 崩溃；
未 ACK；
duplicate processing；
retry；
Stream trim。

再和 PostgreSQL JobRuntime 比：

durability
claim model
retry
idempotency
transaction boundary
recovery
observability
operational complexity

Prediction

[!question]

Pub/Sub 和 Stream 最大的 delivery model 区别是什么？

Worker 收到 Stream message 后立刻执行外部副作用，但 ACK 前崩溃，会发生什么？

为什么 XACK 不能自动保证业务 exactly-once？

BodySense 已有 PostgreSQL JobRuntime 时，Redis 最小增量价值是什么？

DB commit 成功、Redis wakeup 失败，系统怎样保证 job 最终仍能运行？

[!success]- Answer / 解析 1. Pub/Sub 与 Stream 最大的 delivery model 区别是什么？
Pub/Sub 是在线瞬时广播，Subscriber 离线通常直接错过；Stream 会保留 Entry，可以按 ID 读取历史，还提供 Consumer Group、Pending 与 ACK，因此能够保存消费进度。

2. Worker 完成外部副作用但 ACK 前崩溃，会发生什么？
消息仍然处于 Pending / 未确认状态，之后可能被重新 claim 或重新处理。外部副作用却已经发生，因此下一次处理会面对 duplicate delivery。

3. 为什么 XACK 不能自动保证业务 exactly-once？
ACK 只说明 Redis 认为这条消息已经被消费，不可能把 Redis 的 ACK 与数据库写入、HTTP 调用、支付等外部副作用天然组成一个原子事务。side effect success → crash → no ACK 就足以产生重复，因此还需要 idempotency / unique key / transactional boundary。

4. BodySense 已有 PostgreSQL JobRuntime 时，Redis 的最小增量价值是什么？
先作为低延迟 wakeup / notification substrate，减少 Worker 的周期性 DB polling 等待；Job、attempt、legal transition、idempotency、recovery 等 authority 继续留在已有 PostgreSQL Runtime。

5. DB commit 成功、Redis wakeup 失败，怎样保证 Job 最终运行？
Worker 仍定期执行 PostgreSQL recovery / ClaimPending sweep。Redis 只缩短“有新任务到达 → Worker 开始 claim”的延迟，不负责最终可达性，因此丢 wakeup 最多增加等待时间，不应永久丢 Job。

L4 Gate

能解释：

为什么“Redis 唤醒 + PostgreSQL Job authority”在当前 BodySense 中比直接迁移全部任务到 Redis queue 风险更低？