在现代互联网应用中，实时数据同步和事件驱动架构变得越来越重要。无论是订单状态推送、消息通知，还是多系统间的数据一致性，开发者都需要一种高效、可靠的方式来监听数据库变更。MongoDB Change Streams 正是为此而生的强大工具。本文将详细介绍 Change Streams 的原理、用法、常见问题与最佳实践，帮助你在实际项目中高效落地[1]。

Change Streams 是什么？

Change Streams 是 MongoDB 3.6 版本引入的重要特性，它允许应用程序实时监听数据库中的变更事件。与传统的轮询方式相比，Change Streams 具有以下优势[2]：

  • 实时性：变更发生后立即通知应用程序
  • 可靠性：基于复制集的操作日志（oplog）实现，确保不会遗漏任何变更
  • 灵活性：支持过滤和聚合操作，可以精确控制需要监听的变更类型
  • 一致性：保证变更事件的顺序性和完整性

传统方案的局限性

在 Change Streams 出现之前，常见的变更监听方案有：

  1. 轮询（Polling）：定期查询数据库检查变更。这种方式实现简单，通常通过记录上次查询的时间戳，然后定期查询 updatedAt 字段大于该时间戳的文档。优点是易于实现，缺点是实时性差，查询间隔难以权衡，频繁轮询会增加数据库负载，且容易遗漏高并发下的变更，存在时钟同步和并发更新问题。
  2. Tailable Cursors：利用 MongoDB 的 tailable cursor 功能，只能用于 capped collections（定长集合），常见于日志、消息队列等场景。优点是可以持续获取新数据，缺点是 capped collections 不能用于普通业务表，且不支持复杂过滤，游标超时后需要重建，维护复杂。
  3. 应用层触发器：在应用层实现业务逻辑触发器，比如在写操作后主动推送消息或调用回调。优点是灵活性高，缺点是需要开发者自行维护触发逻辑，分布式环境下同步难度大，容易引入一致性和性能问题，维护成本高。
  4. 直接读取 Oplog：通过访问 MongoDB 复制集的操作日志（oplog.rs）来捕获所有数据库操作。优点是可以获取所有变更，缺点是需要复制集权限，oplog 结构复杂，需自行处理轮转、断点续传、权限和安全等问题，开发和运维门槛高。

基于 Oplog 的实现示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
const mongoose = require('mongoose');

async function watchOplog() {
    await mongoose.connect('mongodb://localhost:27017,localhost:27018,localhost:27019/test'); // 需复制集
    const oplog = mongoose.connection.db.collection('oplog.rs');
    const lastTimestamp = await oplog.findOne({}, {}, { sort: { $natural: -1 } });
    
    const cursor = oplog.find({ ts: { $gt: lastTimestamp.ts } })
        .tailable({ awaitData: true });
    
    while (await cursor.hasNext()) {
        const doc = await cursor.next();
        // 处理变更
        console.log('操作类型:', doc.op);
        console.log('命名空间:', doc.ns);
        console.log('操作数据:', doc.o);
    }
}

Change Streams 的工作原理与优势

Change Streams 基于 MongoDB 复制集机制，通过监听 oplog 捕获变更，并将其转换为标准化事件。相比直接 tail oplog，Change Streams 提供了更友好的 API、更好的可靠性和安全性，以及更完善的功能[3]。

变更事件包含：

  • 操作类型（insert、update、delete 等）
  • 文档的完整状态
  • 变更的详细信息
  • 时间戳

基础用法与代码示例

下面是一个使用 Mongoose 实现 Change Streams 的基础示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
const mongoose = require('mongoose');

// 定义 Schema
const userSchema = new mongoose.Schema({
  name: String,
  email: String,
  status: String
});

const User = mongoose.model('User', userSchema);

async function watchCollection() {
  await mongoose.connect('mongodb://localhost:27017,localhost:27018,localhost:27019/test'); // 需复制集
  const changeStream = User.watch();
  changeStream.on('change', (change) => {
    // 监听到变更事件
    console.log('变更类型:', change.operationType);
    console.log('变更文档:', change.fullDocument);
  });
  
  // 错误处理
  changeStream.on('error', (error) => {
    console.error('Change Streams 错误:', error);
  });
}

watchCollection().catch(console.error);

进阶用法：过滤、断线重连与聚合

过滤特定变更

可以使用聚合管道过滤需要监听的变更：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
const pipeline = [
  {
    $match: {
      $or: [
        { operationType: 'insert' },
        { operationType: 'update', 'fullDocument.status': 'active' }
      ]
    }
  }
];
const changeStream = User.watch(pipeline, { fullDocument: 'updateLookup' });
changeStream.on('change', (change) => {
  // 只处理 insert 或 status 为 active 的 update
  console.log('变更:', change);
});

这里需要特别注意，Change Streams 变更事件中的 fullDocument 字段是一个完整的文档对象，因此在 $match 过滤时，必须使用点号表达式（如 fullDocument.status）来匹配内部字段。如果直接使用嵌套对象（如 fullDocument: { status: 'active' }），是无法正确过滤到对应变更事件的，这是很多初学者容易踩的坑。

断线重连策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
let changeStream;
let resumeToken;
async function startStream() {
  try {
    // 使用 resumeToken 恢复监听
    const options = resumeToken ? { resumeAfter: resumeToken } : {};
    changeStream = User.watch([], options);
    changeStream.on('change', (change) => {
      resumeToken = changeStream.resumeToken; // 保存最新 resumeToken
      console.log('变更:', change);
    });
    changeStream.on('error', async (error) => {
      console.error('错误:', error);
      await new Promise(resolve => setTimeout(resolve, 5000));
      await startStream();
    });
  } catch (error) {
    console.error('启动 Change Streams 失败:', error);
    await new Promise(resolve => setTimeout(resolve, 5000));
    await startStream();
  }
}
await startStream();

这里的 resumeToken 实际上就是 change 事件中的 _id 字段，它用于标识变更事件的唯一性，因此，你也可以在数据同步等场景中应用。

原理分析

针对上面提到的过滤特性，我们可以通过 db.currentOp() 命令来查看当前的 Change Streams 操作，下面是一个示例（我删除了一些字段），可以看到它实际上是用 getmore 命令来获取变更事件的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
{
  "type": "op",
  "op": "getmore",
  "ns": "test.users",
  "command": {
    "getMore": {
      "low": -362770267,
      "high": 966127719,
      "unsigned": false
    },
    "collection": "users",
    "batchSize": 1000,
    "$db": "test"
  },
  "planSummary": "COLLSCAN",
  "cursor": {
    "tailable": true,
    "awaitData": true,
    "originatingCommand": {
      "aggregate": "users",
      "pipeline": [
        {
          "$changeStream": {
            "fullDocument": "updateLookup"
          }
        },
        {
          "$match": {
            "$or": [
              {
                "operationType": "insert"
              },
              {
                "operationType": "update",
                "updateDescription.updatedFields.status": {
                   "$exists": true
                 }
              }
            ]
          }
        }
      ],
    }
  }
}

实际应用案例: 实时数据同步

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
// 源数据库模型
const sourceUserSchema = new mongoose.Schema({
  name: String,
  email: String,
  status: String
});

const SourceUser = mongoose.model('SourceUser', sourceUserSchema);

// 目标数据库模型
const targetUserSchema = new mongoose.Schema({
  name: String,
  email: String,
  status: String
});

const TargetUser = mongoose.model('TargetUser', targetUserSchema);

async function syncData() {
  await mongoose.connect('mongodb://localhost:27017,localhost:27018,localhost:27019/test'); // 需复制集
  const changeStream = SourceUser.watch([], { fullDocument: 'updateLookup' });
  changeStream.on('change', async (change) => {
    switch (change.operationType) {
      case 'insert':
        await TargetUser.create(change.fullDocument);
        break;
      case 'update':
        await TargetUser.findOneAndUpdate(
          { _id: change.fullDocument._id },
          change.fullDocument,
          { upsert: true, new: true }
        );
        break;
      case 'delete':
        await TargetUser.findByIdAndDelete(change.documentKey._id);
        break;
    }
  });
}

syncData().catch(console.error);

常见问题、最佳实践与注意事项

  • 监听不到变更？

    • 检查 MongoDB 是否为复制集或分片集群模式，单节点部署无法使用 Change Stream。
    • 检查连接字符串，确保包含所有副本集节点地址，推荐使用 MongoDB 官方 URI 格式。

  • 断线恢复与数据丢失风险

    • Change Streams 支持断线重连，可通过 resumeToken 实现断点续传。建议每次变更事件都持久化最新的 resumeToken，如存储到 Redis、文件或数据库，避免进程重启后丢失。
    • oplog（操作日志）有保留时间限制（通常为若干小时到几天，取决于实例配置和写入压力），断线时间超过保留期将无法恢复，需重新全量同步。建议定期通过 rs.printReplicationInfo() 检查 oplog 大小和时间范围，业务高峰期可适当增大 oplog 保留时长，以防升级或故障期间丢失数据[4]。

  • 性能与资源消耗

    • 每个 Change Streams 都会占用服务器连接和资源。建议合理设置过滤条件（如 $match），只监听关心的事件，减少无关数据流入。
    • 控制 Change Streams 的数量，避免在同一应用或服务中开启过多监听，防止连接池耗尽，影响数据库整体性能。
    • Change Streams 无法利用索引，过多的高并发监听会影响数据库性能，建议合并监听需求或采用更粗粒度的监听。[4:1]
    • 对于高并发写入场景，不建议用 Change Streams 做全量同步，可结合消息队列（如 Kafka）或批量同步方案。

  • 数据一致性与顺序性

    • Change Streams 能保证单集合内事件顺序。对于分片集群，MongoDB 通过全局逻辑时钟保证所有分片变更的全局顺序，但如果某些分片长时间无活动（cold shard），可能导致响应延迟增加。可通过调整 periodicNoopIntervalSecs 降低冷分片延迟[4:2]。
    • 业务侧如需强一致性，需结合业务主键、时间戳等机制做幂等处理。

  • 文档大小与事件限制

    • Change Streams 返回的单个变更事件必须小于 16MB BSON 限制。对于大文档的 insert/replace/update，若事件超出限制会导致通知失败。MongoDB 6.0.9+ 可用 $changeStreamSplitLargeEvent 聚合阶段拆分大事件[4:3]。

  • 集合/数据库 drop 或 rename 行为

    • 如果监听的集合或数据库被 drop 或 rename，相关 Change Streams 会在 oplog 推进到该操作时自动关闭。此时客户端需重新建立监听。
    • 使用 fullDocument: updateLookup 时，若文档已被删除，返回的 fullDocument 可能为 null。

  • 监控与安全

    • 生产环境建议配合监控系统（如 Prometheus、Grafana）监控 Change Streams 的运行状态、延迟和异常。
    • 注意权限分配，避免监听敏感集合或暴露敏感数据。

  • 最佳实践总结

    • Change Streams 适合事件驱动、实时同步、数据驱动通知等场景。
    • 断线重连时务必持久化 resumeToken，并做好异常重试和告警。
    • 结合消息队列可实现更复杂的事件分发和解耦。
    • 充分利用聚合管道过滤，提升性能和可维护性。

结语

Change Streams 为 MongoDB 带来了强大的实时数据监听能力，极大简化了数据同步和事件驱动架构的实现。合理使用 Change Streams，可以提升系统的实时性和可靠性，但也需要关注其性能和资源消耗。建议在实际项目中结合业务需求和资源情况，选择合适的实现方式。

参考资料

  1. MongoDB Change Streams Documentation ↩︎

  2. MongoDB Change Events Reference ↩︎

  3. MongoDB 4.2 内核解析 - Change Stream ↩︎

  4. MongoDB Change Streams Production Recommendations ↩︎ ↩︎ ↩︎ ↩︎