变更流能做什么
watch() 让应用订阅数据变更：集合级、库级或整个实例级都可以监听，事件里带操作类型、文档主键与完整文档（可选）。它基于 oplog 实现，因此天然有序、支持断点续传——把 resume token 存下来，重启后接着上次位置继续消费。这正是搜索索引同步、缓存失效与事件驱动链路常用的基础设施。
基本信息
| 前提 | 复制集或分片集群（单节点非复制集不支持） |
|---|---|
| 语法 | db.coll.watch(pipeline, options)，也可 watch 整个库或实例 |
| 事件字段 | operationType、documentKey、fullDocument、updateDescription、clusterTime |
| 断点续传 | resumeAfter、startAfter 或 startAtOperationTime 指定起始位置 |
| 完整文档 | fullDocument: "updateLookup" 可取当前已提交版本 |
| 权限要求 | 需要集合或库级别的变更流读取权限 |
典型使用场景
- 搜索索引同步：把变更投递给索引更新链路，替代定时全量扫描
- 缓存失效：文档更新后主动清理或刷新对应缓存键
- 事件驱动：订单状态变化触发通知、风控与下游流程
- 审计与留痕：把变更事件写入专用集合供追溯
- 数据迁移过渡期：新旧结构并存时同步写入两套存储
mongosh 示例
mongosh 示例
// 只关注插入事件，并只输出必要字段
const cs = db.orders.watch([
{ $match: { operationType: { $in: ["insert", "update"] } } },
{ $project: { "documentKey._id": 1, operationType: 1, fullDocument: 1, updateDescription: 1 } }
])
cs.on("change", (ev) => {
printjson({ key: ev.documentKey, op: ev.operationType, doc: ev.fullDocument, token: ev._id })
})
// 应用侧持久化 ev._id 作为 resume token：重启后用 resumeAfter 续传
// 更新事件默认不带完整文档，需要时在 watch 选项里指定 fullDocument 读取当前版本事件字段读法
| 字段 | 含义 | 使用要点 |
|---|---|---|
| operationType | insert / update / replace / delete / invalidate 等 | invalidate 表示事件已失效，需要重新建立监听 |
| documentKey | 被操作文档的主键 | 作为幂等去重与目标定位的依据 |
| fullDocument | 操作后的完整文档 | 更新事件默认不返回，需显式指定读取方式 |
| updateDescription | 更新的字段与删除的字段清单 | 用于只同步变化部分，减少写入量 |
| clusterTime | 集群时间戳 | 可作为顺序参考与调试依据 |
| _id（事件 id） | resume token | 必须持久化保存才能断点续传 |
变更流受 oplog 保留窗口约束：消费者长时间断开后如果起始位置已被截断，就只能改从当前时间重新开始，因此 token 与消费延迟都要监控。
变更事件是至少一次投递：消费端必须幂等，重复收到同一事件时不能重复产生副作用。