生态集成

这一课你将能做到

1. 用 Change Streams 监听 MongoDB 数据变更，并说明 resume token 和容错机制
2. 解释 MongoDB Connector for Kafka 的双向同步逻辑，以及 MongoDB + Elasticsearch 的索引同步策略
3. 判断一个场景适合 Change Streams、Kafka Connector 还是定时批量同步

为什么要学

MongoDB 不是孤岛：用户提交代码后，需要把事件发给 Kafka → 消费服务生成排行榜；搜索需要把课程标题同步到 Elasticsearch；运营后台需要把行为日志流进数据仓库。这些链路都依赖「数据从 MongoDB 流出去」或「数据从外部流进 MongoDB」的桥接能力。

Change Streams 是「MongoDB 原生的 push 流」，Kafka Connector 是「企业级桥梁」，ES 同步是「搜索引擎补强」。这一课把三者讲清楚，让你知道哪个场景用哪把刀。

先建立图像

三种集成模式：

① Change Streams（本课重点）
   MongoDB 写入
        │
   oplog ──> Change Stream 监听 ──> 应用消费
        │                      （resume token 断点续传）
   典型：站内消息推送、实时通知、缓存失效

② Kafka Connector（CDC：Change Data Capture）
   MongoDB ──> Source Connector ──> Kafka Topic ──> 消费方
   消费方 ──> Sink Connector ──> MongoDB（反向写回）
   典型：跨系统事件总线、数据仓库同步

③ ES 同步（搜索增强）
   MongoDB 写入 ──> 应用双写或 connector ──> Elasticsearch
   查询走 ES，详情走 MongoDB
   典型：站内搜索、全文检索、聚合分析

跟着做

步骤 1：Change Streams 基础监听

条件：必须是复制集或分片集群（单机无 oplog，不支持）。

// mongosh 里直接监听
db.submissions.watch([]).on("change", doc => printjson(doc))

// 用特定过滤：只监听 accepted 状态的新插入
const pipeline = [
  { $match: { "fullDocument.status": "accepted", operationType: "insert" } }
]
db.submissions.watch(pipeline).on("change", doc => {
  print(`新提交: user=${doc.fullDocument.userId} lesson=${doc.fullDocument.lessonSlug}`)
})

插入一条测试：

db.submissions.insertOne({ userId: 1024, lessonSlug: "mdbrm-integrate", status: "accepted" })

Change Stream 立即收到事件：operationType: "insert"，ns 是库名.集合名，fullDocument 是插入的完整文档。

步骤 2：Resume Token —— 断点续传

// 启动时记录 token，崩溃后从 token 恢复
const watcher = db.submissions.watch([])
watcher.on("change", doc => {
  const token = doc._id   // resume token
  // 持久化到 Redis / 文件 / 另一个集合
  db.resumeTokens.updateOne({ name: "submissions" }, { $set: { token } }, { upsert: true })
})

// 恢复时读取持久化的 token
const saved = db.resumeTokens.findOne({ name: "submissions" })
const resumeAfter = saved ? { _data: saved.token._data } : undefined
db.submissions.watch([], { resumeAfter })