第24章：Mongo Change Streams 实战——订单变革实时推送

1. 项目背景

业务场景：本地生存电商的用户诉苦——"我下完单不知道什么时候发货，每次都要手动刷新订单页面。""优惠券快过期了也不提醒，白白浪费了。" 产品经理提出"订单状态变更实时推送"需求——用户下单后、支付成功、商家接单、骑手取货、订单完成，每个状态变更都要实时推送到 App。同时，数据部门要求订单状态变革后，同步刷新 Redis 缓存、更新 Elasticsearch 搜索索引、发送 Kafka 消息给风控系统。
开辟的第一反应是——"在每个状态的更新接口里直接发消息不就行了？" 但这样会紧耦合：订单服务必要在代码里硬编码"改完状态后通知缓存、搜索、风控"等多个下游，任何一个下游挂掉都会影响订单状态的正常更新。用 Change Streams 可以解耦——订单服务只管改数据库，下游系统监听数据库的变更事件，各自治地响应。
痛点：不用 Change Streams 的下场——紧耦合的服务间调用，一点故障扩散到全局；状态变更通知遗漏或重复（消息没确认、网络抖动、服务重启）；Kafka/Redis/ES 等下游有各自的消费节奏，耦合在一起很难做背压。
2. 项目设计

小胖（手机震个不停）：大师！我把订单状态改完后直接在代码里调用 4 个下游服务——缓存刷新、搜索索引同步、消息推送、风控上报。结果风控服务一挂，整个订单状态更新接口全部超时！
大师：这是紧耦合的典范问题——订单服务不应该知道谁关心订单变革。正确的架构是——订单服务"喊一声"数据库变更，让关心的人自己去听。
小胖："喊一声"？怎么喊？
大师：用 MongoDB 的 Change Streams。它基于复制集的 Oplog，将数据库的任何变更（Insert/Update/Delete/Replace）事件以实时流的形式暴露给应用层。你的订单服务只管写数据库，下游的缓存服务、搜索服务、消息推送各自 watch 订单集合的变更流，独立处理自己的逻辑。任何下游挂了不影响订单写入。
技术映射：Change Streams 是 MongoDB 的事件驱动架构基础设施——它把 Oplog 的变更事件转化为应用层可消费的流式 API。每个事件包罗 operationType（insert/update/delete…）、fullDocument（变更后的完整文档）、ns（发生变更的集合）等信息。
小胖：那这不是跟 Kafka 很像？为什么不用 Kafka？
大师：补集关系，不是替代。Change Streams 负责数据库到应用的实时变更事件流——零代码侵入，数据库操作自动产生事件。Kafka 负责应用到应用的异步消息——你可以在消费 Change Streams 事件后，再投递到 Kafka 给更远的下游。两者组成事件驱动架构的管道。
技术映射：Change Streams = 数据库的 CDC（Change Data Capture）。Kafka = 应用间消息队列。Change Streams → Kafka Connector（如 Debezium）是常见的大数据管道模式。
小白（追问）：Change Streams 基于 Oplog，那它和直接读 Oplog 有什么不同？是不是更安全？
大师：直接读 Oplog 是内部实现细节——MongoDB 版本升级大概改变 Oplog 格式，你的代码就得跟着改。Change Streams 是官方稳定的 API——抽象掉了 Oplog 的内部布局，提供 resumeToken 做断线续听、fullDocument 选项控制回调内容、pipeline 做事件过滤。
小白：那假如监听的客户端断线了，重连之后会不会丢事件？
大师：不会——只要你的 Oplog 窗口足够大，Change Streams 可以回溯到断线前的位置。每个事件都有 _id（resumeToken），客户端记录最后一次成功处理的 token，重连时从这个 token 之后继续。这就是 At-Least-Once 语义——事件大概重复但不会丢失，消费端必须做幂等处理。
大师（总结）：Change Streams 三个要点——① 数据库变更自动变为事件流，解耦写入和下游；② resumeToken 实现断线续听不丢事件；③ 配合聚合管道可以过滤事件（只监听 status 变革而非所有字段变革）。
3. 项目实战

3.1 情况预备

Change Streams 必要复制集情况（单机不支持 Oplog，也就没有 Change Streams）。用第 17 章的 3 节点复制集。

1. docker compose -f mongodb-lab/replicaset/docker-compose-rs.yml up -d

复制代码

3.2 分步实现

步骤一：启动 Change Streams 监听
目的：在 mongosh 中监听订单表的变更事件。

1. // 在 Primary 上连接的 mongosh
2. use local_life
3. db.orders_cs.drop()
5. // 开启 Change Stream
6. const pipeline = [
7.   {
8.     $match: {
9.       // 只监听 status 相关的 update
10.       $or: [
11.         { operationType: "insert" },
12.         { operationType: "update", "updateDescription.updatedFields.status": { $exists: true } },
13.         { operationType: "replace" }
14.       ]
15.     }
16.   }
17. ]
19. const changeStream = db.orders_cs.watch(pipeline, {
20.   fullDocument: "updateLookup"  // update 事件中返回完整的更新后文档
21. })
23. print("Change Stream 已启动，等待事件...")
25. // 非阻塞方式取下一个事件（mongosh 中 tryNext 不会阻塞）
26. let event
27. const maxEvents = 10
28. let count = 0
29. function processNext() {
30.   event = changeStream.tryNext()
31.   if (event) {
32.     count++
33.     print(`\n=== 事件 ${count} ===`)
34.     print("  操作类型:", event.operationType)
35.     print("  集合:", event.ns.coll)
36.     print("  文档ID:", event.documentKey?._id || event.documentKey)
37.     if (event.fullDocument) {
38.       print("  完整文档:", JSON.stringify(event.fullDocument).slice(0, 150))
39.     }
40.     if (event.updateDescription) {
41.       print("  变更字段:", JSON.stringify(event.updateDescription.updatedFields))
42.     }
43.     print("  resumeToken:", event._id._data?.slice(0, 30) + "...")
44.   }
45.   return event
46. }
48. // 初始检测
49. processNext()

复制代码

步骤二：触发各种变更事件
目的：插入、更新、替换订单，观察 Change Stream 捕获到的事件。

1. // 在另一个 mongosh 窗口连接到 Primary
2. use local_life
4. // 1. 插入订单 → 触发 insert 事件
5. const insertResult = db.orders_cs.insertOne({
6.   orderNo: "CS_INSERT_001",
7.   userId: "U_001",
8.   status: "待支付",
9.   totalAmount: NumberDecimal("299.00"),
10.   createdAt: new Date()
11. })
12. print("插入:", insertResult.insertedId)
14. // 2. 更新状态 → 触发 update 事件（仅当我们监听 status 时）
15. db.orders_cs.updateOne(
16.   { orderNo: "CS_INSERT_001" },
17.   {
18.     $set: { status: "已支付" },
19.     $currentDate: { paidAt: true }
20.   }
21. )
22. print("更新状态: 待支付 → 已支付")
24. // 3. 更新其他字段（不涉及 status）→ 不触发事件（被 pipeline 过滤了）
25. db.orders_cs.updateOne(
26.   { orderNo: "CS_INSERT_001" },
27.   { $set: { note: "这是备注" } }
28. )
29. print("更新非status字段 → 应被过滤")

复制代码

1. // 回到 Change Stream 窗口执行 processNext() 查看捕获的事件
2. while (processNext()) {}

复制代码

步骤三：实现完整的事件消费循环（模拟后端服务）
目的：模拟一个消费者——订单状态变更后同步 Redis 缓存。

1. // === 模拟 Redis 缓存刷新消费者 ===
2. // 这个模式展示了如何在实际后端代码中使用 Change Streams
4. function startCacheSyncWorker(collectionName, maxRunSecs = 30) {
5.   print("缓存同步 Worker 启动...")
7.   const pipeline = [
8.     {
9.       $match: {
10.         operationType: { $in: ["insert", "update", "replace"] },
11.         "fullDocument.status": { $exists: true }
12.       }
13.     }
14.   ]
16.   const stream = db.getCollection(collectionName).watch(pipeline, {
17.     fullDocument: "updateLookup"
18.   })
20.   let processedCount = 0
21.   let lastResumeToken = null
22.   const startTime = Date.now()
24.   while (Date.now() - startTime < maxRunSecs * 1000) {
25.     const event = stream.tryNext()
26.     if (event) {
27.       processedCount++
28.       lastResumeToken = event._id
30.       // 模拟同步到 Redis
31.       const cacheKey = `order:${event.fullDocument.orderNo}`
32.       print(`  [Redis] SET ${cacheKey} = ${event.fullDocument.status}`)
34.       // 模拟发送 App 推送
35.       if (event.operationType === "update" && event.updateDescription?.updatedFields?.status) {
36.         const newStatus = event.updateDescription.updatedFields.status
37.         print(`  [Push] 订单 ${event.fullDocument.orderNo} 状态变更为: ${newStatus}`)
38.       }
39.     } else {
40.       sleep(500)  // 空闲时休眠 500ms
41.     }
42.   }
44.   stream.close()
45.   print(`Worker 停止: 处理了 ${processedCount} 个事件`)
46.   return { processedCount, lastResumeToken }
47. }
49. // 启动 worker（在 mongosh 中用非阻塞模式运行几秒）
50. const result = startCacheSyncWorker("orders_cs", 10)
51. print("处理统计:", JSON.stringify(result))

复制代码

步骤四：Resume Token——断线续听
目的：模拟断线后重新毗连并从上次位置继续。

1. // 1. 处理事件并记录 token
2. let lastToken = null
3. const stream1 = db.orders_cs.watch([], { fullDocument: "updateLookup" })
5. // 处理前 3 个事件
6. for (let i = 0; i < 3; i++) {
7.   let event = null
8.   // 等待事件
9.   while (!event) {
10.     event = stream1.tryNext()
11.     if (!event) sleep(100)
12.   }
13.   lastToken = event._id
14.   print(`处理事件 ${i + 1}:`, event.operationType, event.fullDocument?.orderNo || event.documentKey)
15. }
16. stream1.close()
17. print("记录最后 token:", lastToken)
19. // 触发更多事件
20. db.orders_cs.insertOne({ orderNo: "CS_AFTER_DISCONNECT", status: "待支付", createdAt: new Date() })
22. // 2. 从 token 恢复（模拟重连）
23. print("\n从 token 恢复监听...")
24. const stream2 = db.orders_cs.watch([], {
25.   fullDocument: "updateLookup",
26.   resumeAfter: lastToken    // 从这个 token 之后开始
27. })
29. let recovered = 0
30. const start = Date.now()
31. while (Date.now() - start < 5000) {
32.   const event = stream2.tryNext()
33.   if (event) {
34.     recovered++
35.     print(`  恢复事件 ${recovered}:`, event.operationType)
36.     if (recovered >= 1) break
37.   } else {
38.     sleep(200)
39.   }
40. }
41. stream2.close()
42. print("恢复处理了", recovered, "个事件 (应 > 0)")

复制代码

步骤五：事件过滤——pipeline 的高级用法
目的：使用更多过滤条件精确监听事件。

1. // 只在"已完成"状态下才触发事件
2. const completedPipeline = [
3.   {
4.     $match: {
5.       $or: [
6.         { operationType: "insert", "fullDocument.status": "已完成" },
7.         { operationType: "update", "updateDescription.updatedFields.status": "已完成" }
8.       ]
9.     }
10.   }
11. ]
13. const filteredStream = db.orders_cs.watch(completedPipeline, {
14.   fullDocument: "updateLookup"
15. })
17. // 插入一条非"已完成"状态的 → 不被监听
18. db.orders_cs.insertOne({
19.   orderNo: "CS_FILTERED_OUT", status: "待支付", createdAt: new Date()
20. })
22. // 更新到"已完成" → 触发
23. db.orders_cs.updateOne(
24.   { orderNo: "CS_FILTERED_OUT" },
25.   { $set: { status: "已完成" } }
26. )
28. const ev = filteredStream.tryNext()
29. print("过滤后收到:", ev ? `${ev.operationType} → ${ev.fullDocument?.status}` : "无事件（可能还未到达）")
30. filteredStream.close()

复制代码

步骤六：复制集 vs 分片集群中 Change Streams 的差别

1. // 1. 在复制集中：Change Stream 在 Primary 上启动
2. //    事件从该节点的 Oplog 产生
4. // 2. 在分片集群中：Change Stream 在 mongos 上启动
5. //    事件从所有分片的 Oplog 合并产生
6. //    mongos 会协调来自多个 shard 的事件流
8. // 查看当前是不是分片集群
9. const isSharded = db.runCommand({ isdbgrid: 1 })
10. print("是否是分片集群:", isSharded.ok === 1 ? "是 (mongos)" : "否 (mongod)")
12. // 在复制集中，Change Stream 连接到的 mongod 如果发生选举（Primary 切换）
13. // 客户端需要重建 Change Stream —— Driver 会自动处理
14. // 参数 resumeAfter / startAfter 可在重连时从断点继续

复制代码

3.3 完整代码清单

文件用途mongodb-lab/replicaset/ch24-change-stream.js基础 Change Stream 监听mongodb-lab/replicaset/ch24-cache-sync.js模拟缓存同步 Workermongodb-lab/replicaset/ch24-resume-token.jsResume Token 断线续听mongodb-lab/replicaset/ch24-pipeline-filter.jsPipeline 事件过滤3.4 测试验证

1. use local_life
3. // 1. 验证 Change Stream 可用（需要复制集环境）
4. try {
5.   const testStream = db.orders_cs.watch([], { fullDocument: "updateLookup" })
6.   print("Change Stream 创建:", "PASS")
7.   testStream.close()
8. } catch (e) {
9.   print("Change Stream 创建:", "FAIL - ", e.message.includes("replica set") ? "需要复制集" : e.message)
10. }
12. // 2. 验证 insert 事件被捕获
13. db.orders_cs.insertOne({ orderNo: "VALIDATE_001", status: "测试", createdAt: new Date() })
14. // 在 Change Stream 消费者中应看到此事件
16. // 3. 验证 resumeToken 断线恢复
17. // 手动记录一个 token，触发事件后，用 resumeAfter 重新连接
18. // 应能看到断线期间的新事件
20. // 4. 验证 pipeline 过滤
21. // 插入 status != 目标值，应不被监听
23. print("\n=== Change Streams 验证完成 ===")

复制代码

4. 项目总结

4.1 Change Streams vs 应用层 Hook vs Kafka CDC

维度Change Streams应用层 Hook（手动发事件）Kafka CDC（Debezium）代码侵入零（数据库侧）高（每个写操作加代码）零事件可靠性高（基于 Oplog）低（异常时易丢）高（Kafka 持久化）延迟< 100ms< 1ms（同步）< 500ms过滤能力内置 pipeline应用层自由Debezium SMT运维复杂度低（MongoDB 内置）无高（需维护 Kafka + Connector）适用场景实时推送、缓存同步事件必须与写入原子绑定时大数据管道、多系统分发4.2 适用场景

Change Streams 适用：

• 订单/物流状态推送——状态变更实时通知用户 App。
  
• Redis 缓存刷新——数据变更后异步更新缓存。
  
• Elasticsearch 索引同步——MongoDB 主存 + ES 搜索引擎的 CDC 管道。
  
• 事件驱动架构的基础设施——其他服务通过 Change Streams 订阅范畴事件。
  
• 实时数据看板——通过 Change Streams 将新数据流式推送至 Dashboard。
  

不适用场景：

• 必要严格的事务绑定（如"发消息必须在同一个数据库事务中成功"）——Change Streams 是异步的、在事务提交后才触发。
  
• 一次性汗青数据的全量导出——Change Streams 只关注增量变更。
  

4.3 注意事项

注意事项说明仅复制集/分片集群支持单节点 mongod 没有 Oplog，不支持 Change StreamsOplog 窗口要够大resumeToken 只能回溯到 Oplog 窗口内，超出则 Change Stream 失效消费端必须幂等Change Streams 提供 At-Least-Once 语义，网络重连大概发送重复事件fullDocument: "updateLookup"假如文档在事件到达前被删除了，fullDocument 为 null不能 watch admin/local/config 库Change Streams 只支持业务数据库的集合4.4 常见踩坑经验

故障案例一：resumeToken 失效后 Change Stream 静默丢失
某团队在 Worker 重启时用旧的 resumeToken 恢复监听，结果发现一批订单的状态变更没被推送到 App。根因：resumeToken 引用的 Oplog 条目已被覆盖——Oplog 窗口太小，重启后的几个小时恢复间隔超出了 Oplog 的保留时长。解决：增大 Oplog 至 24-48 小时；Worker 在每次处理完事件后持久化当前时间戳 + resumeToken，异常重启后查抄"假如时间差过大，改用 startAtOperationTime 而非 resumeAfter"。
故障案例二：fullDocument: "default" 导致 update 事件缺失完整文档
开辟监听 update 事件，只拿到了 updateDescription 而不包罗完整文档，无法判定订单所属用户来发送 App 推送。根因：fullDocument 默认值是 "default"——不返回完整文档。解决：显式设置 fullDocument: "updateLookup"，MongoDB 在事件生成后额外做一次查询返回完整文档（有微小的查询开销），或者通过 documentKey._id 在应用层补查。
故障案例三：分片集群中 change stream 的全局排序问题
某系统在分片集群中用 Change Streams 监听订单变更，期望所有事件按发生时间排序。实际上不保证跨分片的严格时序——shard A 的 10:00:02 的事件大概比 shard B 的 10:00:01 的事件更早到达 mongos。根因：分片集群中 Change Streams 从每个分片独立拉取事件后合并，合并的顺序是"事件到达 mongos 的时间"而非"事件在各自分片上的发生时间"。解决：依赖全局时间的业务逻辑不应依赖 Change Streams 的顺序保证；使用 clusterTime 自行排序和去重。
4.5 思考题

• 假如 Change Stream 消费者处理事件时抛出异常（如 Redis 不可用），应该立即重试、跳过还是停止消费？为什么？
  
• Change Stream 的 startAtOperationTime 和 resumeAfter 有什么区别？在什么场景下用前者而不是后者？
  

（答案将在第 25 章末端揭晓）

上一章思考题答案：

• Double 转 Decimal128：Decimal128 不能直接从 Double 构造（NumberDecimal(3.14) 会把 Double 的浮点误差带进去），需先转为字符串再构造 Decimal128。迁徙脚本：db.collection.aggregate([{$match:{field:{$type:"double"}}}, {$set:{field:{$convert:{input:{$toString:"$field"}, to:"decimal"}}}}, {$merge:{...}}])，使用 $toString 将 Double 转为字符串，再用 $toDecimal 转为 Decimal128——整个过程在聚合管道中原子完成。
  
• 避免老服务因新字段报错：① 新字段用可选类型（Optional），老服务在反序列化时忽略未知字段（如 Jackson 的 @JsonIgnoreProperties(ignoreUnknown=true)）；② 新字段用新名称而非改名——不修改旧字段的类型，只是新增对应的新字段（如 amount_v2: Decimal128 同时保留 amount: Double）；③ 通过版本号灰度，路由层根据请求头或用户 ID 将流量分发到新旧服务。
  

延伸阅读与资源

Redis 8 实战精讲：从 CRUD 到源码，构建高可用缓存系统
Python 3实战精进：从脚本到高并发订单引擎
MongoDB 实战进阶与内核修炼
python入门：Rquests从菜鸟脚本到企业级SDK的网络实战圣经
Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地
后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战
10倍开辟者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开辟之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析