48.5 Watch 与通知机制
一句话
Watch 把「轮询有没有变」变成「变了通知我」；语义上有三种选择——逐事件、按 key 合并、只发版本信号——选错会丢事件、惊群或读到陈旧值；实现上必须回答「断线后从哪续、丢了怎么办」。
三种通知语义
| 语义 | 每次变更发什么 | 优点 | 代价 |
|---|---|---|---|
| 逐事件（per-event） | 一条变更事件 | 不丢中间状态、可做审计 | 突发时事件量 = 变更量 × watcher 数 |
| 按 key 合并（coalesce） | 每个 key 只发最新值 | 事件量小、终态正确 | 中间值丢失，不能做「每次变化都记账」 |
| 版本信号（revision） | 只发「有新版本」 | 最省带宽 | 客户端要自己拉全量/增量，陈旧窗口 = 拉取间隔 |
它们不是实现细节，而是契约：调用方要明确「我需要每一次变化，还是只需要最终状态」。
算一笔账：1 秒 1 万次更新、1000 个 watcher
设定：1000 个 key、1 秒内 1 万次更新（平均每 key 10 次）、1000 个 watcher 都关注全部 key；每个 watcher 的队列容量 1024、每秒能处理 2000 条事件。
package main
import "fmt"
const (
watchers = 1000
updates = 10000
keys = 1000
queueCap = 1024
consume = 2000 // 每个 watcher 每秒能处理的事件数
slots = 100 // 把 1s 切成 100 个 10ms 槽
)
func main() {
perSlot := updates / slots
drain := consume / slots
// 1) per-event：每个事件都进队列，队列满就丢
queue, delivered, dropped, maxQ := 0, 0, 0, 0
for s := 0; s < slots; s++ {
queue += perSlot
if queue > queueCap {
dropped += queue - queueCap
queue = queueCap
}
if queue > maxQ {
maxQ = queue
}
d := drain
if d > queue {
d = queue
}
queue -= d
delivered += d
}
delivered += queue // 突发结束后把积压排空
backlog := queue
// 2) coalesce：每个 key 只保留最新版本，1s 刷一次
coalesced := keys // 1s 内所有 key 都可能变
// 3) revision：只发一个「有新版本」信号，watcher 自己拉全量
signals := 1
fmt.Printf("%-12s %-18s %-14s %-14s %-14s\n", "模式", "送达(全部 watcher)", "丢弃", "最大待发/watcher", "恢复动作")
fmt.Printf("%-12s %-18d %-14d %-14d %-14s\n", "per-event",
delivered*watchers, dropped*watchers, maxQ, "全量重同步")
fmt.Printf("%-12s %-18d %-14d %-14d %-14s\n", "coalesce",
coalesced*watchers, 0, coalesced, "无需（终态正确）")
fmt.Printf("%-12s %-18d %-14d %-14d %-14s\n", "revision",
signals*watchers, 0, 1, "拉全量（≤1s 陈旧）")
fmt.Printf("\n单 watcher：送达 %d，丢弃 %d，突发结束积压 %d，容量 %d\n", delivered, dropped, backlog, queueCap)
// 4) 单个热点 key：1000 watcher × 100 次变更/s
hotChanges := 100
fmt.Printf("热点 key 惊群：per-event %d 通知/s；100ms 合并后 %d 通知/s\n",
hotChanges*watchers, (1000/100)*watchers)
}
输出：
模式 送达(全部 watcher) 丢弃 最大待发/watcher 恢复动作
per-event 3004000 6996000 1024 全量重同步
coalesce 1000000 0 1000 无需（终态正确）
revision 1000 0 1 拉全量（≤1s 陈旧）
单 watcher：送达 3004，丢弃 6996，突发结束积压 1004，容量 1024
热点 key 惊群：per-event 100000 通知/s；100ms 合并后 10000 通知/s
表中 coalesce 的「1000」是待刷新的 key 数，不是事件积压。
三个结论：
- 逐事件在「队列有界 + 消费能力有限」时必丢：单 watcher 只能处理 2000 条/秒，突发 1 万条，队列 1024 只能兜住一部分，70% 被丢；
- 按 key 合并把 1000 万级压到 100 万级，终态仍然正确，但中间值没了；
- 版本信号最省（1000 条通知/秒），代价是客户端要自己拉、且最多陈旧一个拉取周期。
🔑 选择语义的依据是业务需要中间态还是终态：审计/计费需要每一次变化，缓存/配置只需要终态。把需要终态的业务做成逐事件，是惊群和丢事件的常见来源。
顺序与重复
| 保证 | 含义 | 怎么用 |
|---|---|---|
| 单 key 顺序 | 同一 key 的事件按版本递增 | 用版本号丢弃旧事件（if rev <= lastRev { skip }） |
| 全局顺序 | 所有事件有单调 revision/zxid | etcd revision、ZK zxid、Consul index |
| 至少一次 | 断线重连/重试可能重复投递 | 处理函数必须幂等 |
| 不保证「恰好一次」 | watch 不是事务日志 | 不要用它做账本/消息队列 |
断线续传：从哪个 revision 接着看
这是 watch 设计里最关键的问题。四种常见答案：
| 产品 | 续传方式 | 丢事件怎么办 |
|---|---|---|
| etcd | watch 带 start_revision，可回放历史（在 compaction 窗口内） | 版本被压缩 → ErrCompacted → 全量 LIST 后重新 watch |
| ZooKeeper | 标准 watch 是一次性的，触发后要重新注册 | 重新注册前先 get 一次拿最新值；持久 watch（3.6+）减少重注册 |
| Consul | blocking query 带 index，值变了或超时返回 | index 回退/重置时全量拉一次 |
| Nacos | 长轮询带 MD5，服务端比对后返回 | MD5 不一致就拉全量 |
| K8s informer | LIST + WATCH，带 resourceVersion | 410 Gone / 过期 → 重新 LIST，重建缓存 |
通用的「list + watch」模式：
1. LIST 全量，记下 revision R
2. WATCH from R
3. 收到事件 → 更新本地缓存 + 处理
4. 断线 → 重新 LIST（或 WATCH from lastRev）
5. 如果 revision 已被压缩 → 必须重新 LIST
⚠️ 只 watch 不 resync 是错的。watch 是「变更提示」，不是「完整状态」；任何一次丢事件、队列溢出、版本压缩，都需要一次全量对账来恢复。K8s 的 informer 之所以有 resync 周期，就是这个原因。
实现方式对照
| 产品/机制 | 模型 | 特点 | 坑 |
|---|---|---|---|
| etcd watch | 流式、按 revision | 可回放、prefix watch、progress notify | 慢 watcher 可能积压或被断开；compaction 后要重 LIST |
| ZooKeeper watch | 一次性（标准）/持久（3.6+） | 会话内有序；连接事件也会通知 | 重注册间隙丢事件；watch 数量受内存限制 |
| Consul blocking query | 长轮询 index + wait | 无推送、兼容 HTTP | 每个 watcher 占一个连接；要处理 index 回退 |
| Nacos 长轮询 | 客户端带 MD5，服务端 hold | 配置/注册都能用 | 30s 超时后要重发；变更风暴时合并 |
| K8s informer | LIST+WATCH+本地缓存 | 工业级标准实现 | resourceVersion 过期要重建；handler 不能阻塞 |
断线续传的验收：一段能跑的脚本
E="--endpoints=127.0.0.1:2379"
# 1) 记住上次处理到的 revision（生产上必须持久化）
REV=$(etcdctl $E get /config/order/retry -w json \
| python3 -c 'import json,sys; print(json.load(sys.stdin)["header"]["revision"])')
# 2) 从 revision+1 开始 watch；prefix 表示监听整个前缀
etcdctl $E watch --rev=$((REV+1)) /config/order --prefix
本机 etcd 3.7 实测：
$ etcdctl --endpoints=127.0.0.1:2379 put /config/order/timeout 2s
OK
$ etcdctl --endpoints=127.0.0.1:2379 get /config/order/timeout -w json
{"header":{"revision":2,...},"kvs":[{"key":"L2NvbmZpZy9vcmRlci90aW1lb3V0","value":"MnM=","mod_revision":"2"}]}
$ etcdctl --endpoints=127.0.0.1:2379 watch --rev=3 /config/order --prefix
PUT
/config/order/timeout
5s
watch 断开期间写入，重连带 --rev=last+1 后会按顺序补发——不丢、不乱。只有超过 compaction 窗口（见 48.6 产品 的 auto-compaction-retention: "1h"）才回放不了，那时必须重新 LIST 一次。
| 客户端要做的 | 不做的后果 |
|---|---|
| 持久化 last revision | 重启后只能从 0 或最新开始：要么重放全量，要么丢事件 |
| 按 revision 去重 | 重连会重复投递同一事件 |
| 指数退避 + 抖动重连 | 服务端重启时所有客户端同时重连（惊群） |
遇到 ErrCompacted 重新 LIST | 死循环重试，永远补不齐 |
降级：watch 断开 → 指数退避重连（带 last_rev+1）；遇到 ErrCompacted → 全量 LIST 后重新 watch。
什么时候别用：更新频率低于 1 次/分钟、watcher 很少时，轮询 30s 更简单；watcher 规模超过服务端上限要分片或汇总，不要硬扛。
验收三条，缺一不可：
# ① 不丢：断开 → 写入 N 条 → 重连 → 收到 N 条
# ② 不乱：事件按 Header.Revision 单调递增
# ③ 幂等：同一 revision 重复处理，结果不变
惊群与推送风暴
| 场景 | 事件量 | 解法 |
|---|---|---|
| 1000 watcher 关注同一个热点 key，100 次变更/s | 10 万通知/s | 服务端合并（100ms 窗口 → 1 万/s）、按 key 分片、客户端限流 |
| 1000 个实例同时上线 | 每个 watcher 收到 1000 条 ADDED | 增量 + 合并 + 分页；避免全量快照 |
| 所有 follower watch 同一个 leader key | 失效瞬间全员被唤醒 | ZK 只 watch 前驱；etcd 用 Election 封装 |
| 注册中心重启，客户端同时重连 | 连接风暴 | 随机抖动 + 指数退避 + 从 revision 续传 |
💡 惊群的根因不是「watch 太多」，而是「所有 watcher 关注同一个粒度」。把「watch 一个 key」改成「watch 一个范围/前缀 + 本地过滤」，或者让服务端做一次合并再广播，事件量会降一个数量级。
背压与降级
| 策略 | 做法 | 效果 |
|---|---|---|
| 有界队列 | 队列满就断开 watch，要求全量重同步 | 保护服务端与客户端内存；etcd 等实现会对慢 watcher 做断开/重试 |
| 合并窗口 | 100ms 内的同 key 变更只发最后一次 | 事件量降为「变更 key 数」而不是「变更次数」 |
| 批量拉取 | 版本信号 + 定时拉全量/增量 | 适合读多写少、能容忍秒级陈旧 |
| 限流 | 每 watcher/每连接的事件速率上限 | 防止单个客户端拖垮服务端 |
| 降级 | watch 不可用时回退到轮询 | 保证最终可用，牺牲实时性 |
不要静默丢事件：队列溢出时必须让客户端知道（关闭 watch / 返回错误），否则客户端会拿着不完整的缓存继续服务，问题被藏起来。
反模式
| 反模式 | 后果 |
|---|---|
| 把 watch 当消息队列（无持久化、无 ack） | 断线即丢，无法补偿 |
| 断线后只重新 watch，不做全量对账 | 缓存永久缺一块，越跑越偏 |
| 每次事件都全量拉取 | 事件风暴放大成读风暴 |
| 无界队列 / handler 阻塞 | 内存爆、watch 被服务端断开 |
| 所有 watcher watch 同一个热点 key | 惊群，通知量 = 变更 × watcher |
| 依赖 watch 做 exactly-once | 重复/丢失都会发生，业务错账 |
忽略 ErrCompacted / 410 Gone | 客户端永远卡在旧 revision |
| 用 watch 事件里的旧值做决策 | 事件是快照，可能已被后续写覆盖 |
自检
Q1 为什么「逐事件 watch」在突发流量下一定会丢事件？怎么避免？
▸ 答案：事件产生速率受写入驱动，消费速率受客户端处理能力限制；队列容量有限，积压超过容量就只能丢。模拟里 1 秒 1 万次更新、每 watcher 每秒处理 2000 条、队列 1024，最终丢了 6996/10000。避免方式：① 改成按 key 合并或版本信号，把「变更次数」降成「变更 key 数」；② 服务端限流/合并；③ 队列溢出时断开 watch，强制客户端全量重同步，而不是静默丢。
Q2 watch 断线后，为什么不能「从断开的地方继续 watch」就完事？
▸ 答案：因为断开期间可能发生版本压缩（etcd compaction）、会话过期（ZK）、资源版本过期（K8s 410 Gone），旧 revision 已经不可回放；此外队列溢出/客户端重启也会丢事件。正确做法是「LIST 全量 + 记录 revision + 再从 revision watch」，任何一次续传失败都回到全量对账。watch 只保证「变更提示」，不保证「完整历史」。
Q3 什么时候该用「按 key 合并」，什么时候必须「逐事件」？
▸ 答案：只需要终态的场景（缓存失效、配置生效、UI 刷新、路由表更新）用按 key 合并，事件量小且终态正确；需要每一次中间状态的场景（审计日志、计费、事件溯源、按变更次数计费）必须逐事件，并且要接受更高的带宽/存储成本与更复杂的背压。中间态丢失对前者无害，对后者是数据错误。
Q4 ZooKeeper 的标准 watch 有什么坑？持久 watch 解决了什么？
▸ 答案：标准 watch 是一次性的：触发后必须重新注册，注册完成前发生的变更可能看不到；如果客户端在「触发—重注册」之间崩溃/重连，还会丢事件。持久 watch（3.6+）让 watch 在触发后继续有效，减少了重注册间隙，但客户端仍要在会话恢复后做一次全量对账，不能假设「持久 watch = 不丢」。
Q5 1000 个 watcher 关注同一个热点 key，怎么把通知量降下来？
▸ 答案：① 服务端合并（例如 100ms 窗口内只发最后一次，模拟里 10 万/s → 1 万/s）；② 把 watch 粒度从「单个 key」改成「前缀/范围」，客户端本地过滤；③ 按 watcher 分片/分组，由中间层做扇出；④ 客户端限流 + 只保留最新版本；⑤ 如果业务只需要终态，直接改成「版本信号 + 定时拉取」，把推送量降到 watcher 数级别。核心是减少「同一变更 × 多个 watcher」的乘法。