## 01、痛点

凌晨一点，搜索服务开始报错。商品已经下架，搜索结果还能点进去；价格改了，列表页还是旧价格；运营同学截图发到群里，问是不是数据没同步。我是老李，做后端有些年头了，这类问题最后多半回到同一个根上：数据变了，下游没人知道。

产品更新场景很常见：商品中心更新了 Product 集合，搜索侧要刷新索引，缓存要失效，审计系统要记录变更。为了这三件事，很多团队常年用三种办法：定时任务扫表、应用里双写消息、甚至直接读 oplog。定时任务扫表有延迟，扫全表很重，扫增量还要维护时间戳；双写消息要处理跨表事务，商品写成功、消息写失败，数据就不一致；直接 tail oplog 更危险，oplog 格式不是公开稳定接口，升级、权限、滚动都容易踩坑。

为什么现在要讲 Change Streams？因为 MongoDB 官方已经提供了这条链路，但很多人的第一反应还是“它到底靠不靠谱、能不能替代 Kafka”。今天不吹也不黑，把原理、实战和边界一次说清。

## 02、是什么

Change Streams 是 MongoDB 提供的一套订阅数据库变更的接口。应用可以通过 watch() 订阅一个 Collection、一个 Database，甚至整个部署的变化，而无需直接 tail oplog。它背后读的是 oplog，但把 oplog 的原始格式封装成了稳定的变更事件。

使用前有一个硬前提：MongoDB 必须运行在 Replica Set 或 Sharded Cluster 上，单机模式不支持。因为 Change Streams 依赖 oplog，而 oplog 只有副本集才有。同时要求 readConcern 为 majority，保证读到的是已经提交给多数节点的数据。

一次变更会产出一个 Change Event，核心字段大致是：
- _id：这个事件的 Resume Token，用于断点续传。
- operationType：insert、update、replace、delete、invalidate 等类型。
- ns：发生变更的库和集合。
- documentKey：变更文档的 _id。
- fullDocument：变更后的完整文档，需要在 watch 时声明 fullDocument: 'updateLookup'。
- updateDescription：update 事件的具体更新字段。
- clusterTime：变更对应的 oplog 时间。

其中 Resume Token 是重中之重。它记录了事件在 oplog 里的位置。应用重启后，只要把上次处理到的 token 传给 startAfter，就可以从断点继续消费，不用重新扫历史。

## 03、为什么

为什么建议看看 Change Streams，而不是继续用轮询或双写？

第一，实时性和顺序性。定时任务最短也有秒级延迟，而且扫表顺序不稳定；Change Streams 可以做到近实时推送，并且同一条文档的变更按先后顺序到达。

第二，数据可靠性。Change Streams 读的是 majority 提交的数据，减少了读到回滚数据的风险。自己读 oplog 需要处理很多边界条件，而 Change Streams 已经帮你处理了。

第三，接口稳定。oplog 格式在不同版本间可能有变化，直接依赖 oplog 相当于把应用和内部实现绑在一起。Change Streams 是官方公开接口，升级更安全。

第四，订阅粒度灵活。可以从 Collection 到 Database 再到整个部署，适合做统一的数据变更出口。

但也要清楚它的限制。Change Streams 不是无限回放的消息队列，它依赖 oplog 的窗口，历史事件不会永久保留；消费确认、重试、死信这些机制也需要自己实现。这些问题放到后面洞见里再展开。

## 04、架构

从整体看，Change Streams 的链路是：应用 driver 创建 change stream cursor，MongoDB 从 oplog 里读取变更，按条件过滤后把事件推给 cursor，driver 再交给应用消费。可以把它理解成一个封装过的 tailable cursor：游标不会读完就关闭，而是一直等新数据到来。

这里要区分两种消费方式。imperative 方式是传统的同步循环：调 next() 阻塞等待下一条事件。简单直接，但每个游标通常要占住一个线程，资源消耗较大。reactive 方式是基于事件流的非阻塞模型，有新事件时回调处理，线程利用率更高。官方文档在资源消耗方面更推荐 reactive 方式。对高并发、长驻监听的生产场景来说，reactive 通常比 imperative 更合适。

如果是 Sharded Cluster，还需要 mongos 把各个 shard 的事件汇总，按 clusterTime 排序后推给应用。这个顺序是全局近似有序，不是严格的线性一致，但已经能覆盖绝大多数下游同步场景。

理解架构还有一个关键点：Change Streams 是数据库变化的反馈，不是独立的事件总线。它依附于数据库，事件的生命周期受 oplog 窗口限制。这一点直接决定了它和 Kafka 这样的消息队列本质不同。

## 05、实战

下面用 Product 更新触发搜索同步、缓存失效和审计这条链路来演示。代码用 Node.js 的 MongoDB driver 写个最小可运行版本，便于看清流程。

```javascript
const { MongoClient } = require('mongodb');

async function main() {
  const client = new MongoClient('mongodb://localhost:27017/?replicaSet=rs0');
  await client.connect();
  const db = client.db('shop');
  const products = db.collection('products');

  // 监听 products 集合的变化，update/replace 时返回完整文档
  const changeStream = products.watch([], { fullDocument: 'updateLookup' });

  while (await changeStream.hasNext()) {
    const event = await changeStream.next();
    const type = event.operationType;

    if (type === 'insert' || type === 'update' || type === 'replace') {
      const doc = event.fullDocument;
      await refreshSearchIndex(doc);
      await invalidateCache(`product:${doc._id}`);
      await writeAudit(event);
    }

    if (type === 'delete') {
      await removeSearchIndex(event.documentKey._id);
      await invalidateCache(`product:${event.documentKey._id}`);
      await writeAudit(event);
    }
  }
}
```

这段代码是 imperative 写法的示例，适合入门。生产环境建议使用 reactive driver，通过背压控制消费速度，比如 Spring Data MongoDB 里的 reactive stream 或 MongoDB Node driver 的 stream API。

断点续传是生产必备。每次处理完事件后，把 event._id 作为 Resume Token 保存到外部存储。应用重启时，用 startAfter 继续：

```javascript
const resumeToken = JSON.parse(loadResumeToken());
const stream = products.watch([], {
  fullDocument: 'updateLookup',
  startAfter: resumeToken
});
```

需要注意，Change Streams 可能重复投递，所以 refreshSearchIndex、invalidateCache、writeAudit 都要做成幂等。搜索刷新可以用 upsert，缓存失效删除天然幂等，审计写入可以用唯一事件 ID 去重。

订阅 Database 级别同样简单：db.watch() 可以收到库内所有集合的变化，适合做统一数据变更出口。部署级别用 client.watch()，但一般只有平台型场景才需要。

## 06、洞见

很多人会问：Change Streams 能不能直接替代 Kafka？

答案是不能，两者解决的问题并不完全相同。Change Streams 是数据库变化反馈，它天然绑定数据库，解决的是“数据在 MongoDB 里变了，应用怎么第一时间知道”。Kafka 是独立的事件基础设施，解决的是大规模事件的分发、解耦、治理和可靠存储。

用 Change Streams 当 MQ 用，会遇到几个现实问题。第一，回放窗口受 oplog 大小限制，oplog 满一轮后老事件就没了；Kafka 可以按策略保存很久。第二，消费模型弱，单个 change stream 基本是顺序消费，没有成熟的分区、消费组、死信队列和重试机制。第三，跨系统治理能力不足，监控、告警、权限控制、多租户隔离都比 MQ 弱。第四，它没有天然的生产者-消费者解耦，所有消费者都直接对着数据库读，数据库压力会随下游增多而上升。

更合理的定位是组合使用：Change Streams 负责从数据库里准确捕获变化，Kafka 负责把变化可靠地分发给多个下游。常见架构是 Change Streams -> Kafka Connect MongoDB source -> Kafka -> 搜索刷新、缓存失效、审计等消费者。这样数据库变更成为可靠事件源，Kafka 承担事件总线的职责，各自做自己擅长的事。

如果团队规模不大、下游只有一两个、变更频率不高，直接用 Change Streams 处理搜索同步和缓存失效没有问题，不必为了用 MQ 而引入一整条基础设施。但当事件要被多个团队消费、需要长期回放、需要复杂治理时，把它桥接到 Kafka 是更稳的选择。

## 07、行动

落到行动上，Product 更新 -> ChangeStream -> Search 同步这条链可以这样推进。

第一步，确认 MongoDB 版本和部署形态。版本 3.6 以上，且必须是副本集或分片集群。生产环境把 readConcern 设为 majority，这是 Change Streams 能工作的前提。

第二步，选择消费方式。如果是新服务，直接上 reactive driver，设定合理的 batchSize，用背压控制内存。如果团队对 reactive 不熟，先用 imperative 快速验证业务，再逐步切换。

第三步，订阅 products 集合，监听 insert、update、replace、delete 四类事件。update 事件要用 fullDocument: 'updateLookup' 拿到完整文档，否则只有变更字段。

第四步，下游处理做幂等。搜索索引按文档 _id upsert；缓存按 key 删除，必要时延迟双删；审计用事件 _id 做唯一性约束去重。

第五步，保存 Resume Token。每次处理成功后持久化，重启从 startAfter 继续，保证故障恢复后不丢进度。注意：如果处理失败，不要盲目跳过去，否则事件会丢。可以暂停消费或把失败事件记录到本地，人工补处理。

第六步，监控。关注 Change Stream 的消费延迟，可以用 event.clusterTime 与当前时间对比。同时监控 oplog 窗口大小，防止 resume token 过期后再续传失败。

第七步，如果下游开始变多，把 Change Streams 的事件写入 Kafka。用 Kafka Connect 的 MongoDB source connector 是最常见做法，避免自己写桥接程序。之后搜索刷新、缓存失效、审计都可以作为 Kafka 消费者独立扩展。

## 08、小结

Change Streams 是 MongoDB 官方提供的变更订阅接口，适合做数据库变更驱动的近实时同步。它把复杂的 oplog 读取封装成稳定事件，支持 Collection、Database、Deployment 三级订阅，Resume Token 让断点续传变得可行。

但它不是 MQ 的替代品。它解决的是数据库变化反馈，消息队列解决的是事件分发与治理。简单场景可以直接用 Change Streams；需要多消费者、长期回放、复杂治理时，把它接到 Kafka 上更合适。

下次再遇到商品更新后搜索不一致，先别急着加定时任务。把 Product 的变更通过 Change Streams 推给搜索和缓存，很多时候一条链路就能省下大量排查时间。

## 09、链接与概念说明

概念说明：
- Replica Set：MongoDB 的高可用复制结构，包含主节点和多个从节点，数据通过 oplog 异步复制。Change Streams 依赖它。
- oplog：副本集的操作日志，记录所有数据变更。Change Streams 读取它，但应用不应直接依赖它。
- Change Event：Change Streams 推送给应用的事件对象，包含 operationType、documentKey、fullDocument 等字段。
- Resume Token：标识事件在 oplog 中位置的 token，用于断点续传。
- imperative / reactive：两种消费模型。imperative 是阻塞式循环，reactive 是非阻塞事件流。官方推荐 reactive 以减少资源消耗。

官方文档可搜索 MongoDB Change Streams；Kafka 侧可搜索 Kafka Connect MongoDB Source Connector。

凌晨一点，搜索服务开始报错。商品已经下架，搜索结果还能点进去；价格改了，列表页还是旧价格；运营同学截图发到群里，问是不是数据没同步。我是老李，做后端有些年头了，这类问题最后多半回到同一个根上：数据变了，下游没人知道。

《Spring Mongo实战》第15招 ChangeStream做事件流

商品更新后搜索不一致？本文拆解 Change Streams 的原理、实战链路与边界，帮你判断它何时能替代 MQ，何时不能。

架构师

数据库

后端

架构设计

MongoDB ChangeStreams通过订阅oplog提供稳定实时数据变更监听，支持集合、库、部署三级订阅，利用ResumeToken实现断点续传。适合搜索索引刷新、缓存失效、审计等近实时同步场景，但受oplog窗口限制，复杂事件分发需桥接Kafka。本文详解原理、实战代码与架构边界。

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

《Spring Mongo实战》第15招 ChangeStream做事件流-腾讯云开发者社区-腾讯云

《Spring Mongo实战》第15招 ChangeStream做事件流

《Spring Mongo实战》第15招 ChangeStream做事件流

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐