云数据库 MongoDB：使用MongoDB变更流（Change Stream）实时捕获数据变更

监听更多的DDL事件

前提条件

MongoDB 6.0及以上版本（升级指南）。

操作步骤

1. 使用 mongosh 连接数据库。

2. 执行命令watch，设置 showExpandedEvents: true：

   // mongo shell or mongosh v1.x
   cursor = db.getSiblingDB("test").watch([],
     {
       showExpandedEvents: true       // 启用更多DDL 事件监听
     }
   );
   cursor.next();

结果验证

1. 在新的SQL窗口执行变更SQL，如db.createCollection("myCollection1")。

2. 观察原SQL窗口输出上述执行SQL相关信息。

   mg xxx test> cursor = db.getSiblingDB("test").watch([],
   ... {
   ...     showExpandedEvents: true     // 启用 更多 DDL 事件监听
   ... }
   ... );
   ... cursor.next();
   Warning: If there are no documents in the batch, next will block. Use tryNext if you want to check if there are any documents without waiting.
   {
     _id: {
       _data: '8268777614000000012B042C0100296E5A100434919C64814940F3A61F9D97A3F60D5C463C6Fxxx03C63726561746500466F70657274696F6E4465737372697074696F6E0046466964496E6465787800461E76002B04466B657990046E5F6964002B02003C6E616D65003C5F69645F000000004'
     },
     operationType: 'create',
     clusterTime: Timestamp({ t: 1752659476, i: 1 }),
     collectionUUID: UUID('34919c64-8149-40f3-a61f-9xxx'),
     wallTime: ISODate('2025-07-16T09:51:16.043Z'),
     ns: { db: 'test', coll: 'myCollection1' },
     operationDescription: { idIndex: { v: 2, key: { _id: 1 }, name: '_id_' } }
   }

开启前镜像（Pre-image）

MongoDB的pre-image是指文档在修改或删除前的完整状态快照，用于记录数据变更前的原始值。

前提条件

MongoDB 6.0及以上版本（升级指南）。

操作步骤

1. 开启数据库维度前镜像 ：

   db.adminCommand({
     setClusterParameter: {
       changeStreamOptions: {
         preAndPostImages: { expireAfterSeconds: "off" } // "off" 表示使用oplog保留时间
       }
     }
   })

2. 开启集合维度前镜像 ：

   说明

   若需为某数据库内所有集合启用前镜像，则必须先开启数据库维度前镜像，然后在该库的每个集合上单独启用集合级前镜像功能。

   修改现有集合

   db.runCommand({
     collMod: "myCollection",
     changeStreamPreAndPostImages: { enabled: true }
   })

   创建新集合时指定

   db.createCollection("myCollection", { changeStreamPreAndPostImages: { enabled: true }})

3. 建立变更流监听 （指定Pre-image选项）：

   // 在目标集合上建立监听
   cursor = db.getCollection("myCollection").watch([],
     {
       fullDocument: 'required', // 或 'whenAvailable'
       fullDocumentBeforeChange: 'required' // 或 'whenAvailable'
     }
   )
   cursor.next();

   • required: 服务端必须返回Pre/Post-image，否则报错。

   • whenAvailable: 服务端尽力返回，不保证一定有。

结果验证

1. 检查数据库维度的开关是否已开启：

   db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

   开启成功后输出信息。

   test> db.adminCommand( { getClusterParameter: "changeStreamOptions" } )
   {
     clusterParameters: [
       {
         _id: 'changeStreamOptions',
         clusterParameterTime: Timestamp({ t: 1752655937, i: 1 }),
         preAndPostImages: { expireAfterSeconds: Long('100') }
       }
     ],
     ok: 1,
     '$clusterTime': {
       clusterTime: Timestamp({ t: 1752656717, i: 1 }),
       signature: {
         hash: Binary.createFromBase64('xxx=', 0),
         keyId: Long('xxx')
       }
     },
     operationTime: Timestamp({ t: 1752656717, i: 1 })
   }

2. 检查集合配置：

   db.getCollectionInfos({name: "myCollection"}) // 或 db.runCommand({listCollections: 1})

   预期输出：在返回的文档中找到类似 "options" : { "changeStreamPreAndPostImages" : { "enabled" : true } } 的字段。 

   test> db.getCollectionInfos({name: "myCollection"})
   [
     {
       name: 'myCollection',
       type: 'collection',
       options: { changeStreamPreAndPostImages: { enabled: true } },
       info: {
         readOnly: false,
         uuid: UUID('55ed1b7c-7575-4ba3-8afa-xxx')
       },
       idIndex: { v: 2, key: { _id: 1 }, name: '_id_' }
     }
   ]

3. 在另一个mongosh窗口更新 myCollection 中的文档。

4. 观察 cursor 返回的事件，应包含 fullDocumentBeforeChange 字段（变更前文档）。

   ... cursor.next();
   {
     _id: {
       _data: '8268786B15000000012B042C0100296E5A100455ED1B7C75754BA38AFA0C5B56A360BC463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046xxx697CDDF6EE279xxx'
     },
     operationType: 'update',
     clusterTime: Timestamp({ t: 1752722197, i: 1 }),
     wallTime: ISODate('2025-07-17T03:16:37.626Z'),
     fullDocument: {
       _id: ObjectId('6878697cddfxxx'),
       name: 'test',
       count: 111
     },
     ns: { db: 'test', coll: 'myCollection' },
     documentKey: { _id: ObjectId('6878697cddfxxx') },
     updateDescription: {
       updatedFields: { count: 111 },
       removedFields: [],
       truncatedArrays: []
     },
     fullDocumentBeforeChange: { _id: ObjectId('6878697cddf6xxx'), name: 'test' }
   }

更多信息参考Change Streams with Document Pre- and Post-Images。

开启后镜像（Post-image）

MongoDB的Post-image是变更发生后文档的最新完整状态快照，用于记录变更后的完整文档内容。

前提条件

MongoDB 3.6及以上版本（升级指南）。

操作步骤

执行watch命令时设置fullDocument: 'updateLookup'。

cursor = db.getSiblingDB("test").myCollection.watch([], 
  {
    fullDocument: 'updateLookup'
  }
);
cursor.next();
              

结果验证

1. 在另一个mongosh窗口新增或更新 myCollection 中的文档。

2. 观察 cursor 返回的事件，应包含 fullDocument 字段（变更后文档）。

   xxx [primary] test> cursor = db.getSiblingDB("test").myCollection.watch([],
   ... {
   ...     fullDocument: 'updateLookup'
   ... }
   ... );
   ... cursor.next();
   [...
   {
     _id: {
       _data: 'xxx100296E5A10040D6C3FBC28484F08240555576066945463C6F7065726174696F6E54797065003C757064617465500046646F63756D656E744B6579005F6964004B657390046645F696400646881ADE741E7D4B638ED7CA1000004'
     },
     operationType: 'update',
     clusterTime: Timestamp({ t: 1753329166, i: 1 }),
     wallTime: ISODate('2025-07-24T03:52:46.966Z'),
     fullDocument: {
       _id: ObjectId('xxx8ed7ca1'),
       name: 'test1',
       age: 12,
       count: 2222
     },
     ns: { db: 'test', coll: 'myCollection' },
     documentKey: { _id: ObjectId('xxx7ca1') },
     updateDescription: {
       updatedFields: { count: 2222 },
       removedFields: [],
       truncatedArrays: []
     }
   }]

处理超大变更事件 (>16MB)

前提条件

MongoDB 7.0及以上版本（升级指南）。

操作步骤 

在 watch() 的管道中包含 $changeStreamSplitLargeEvent 阶段：

myChangeStreamCursor = db.myCollection.watch(
  [ { $changeStreamSplitLargeEvent: {} } ], // 添加切分阶段
  {
    fullDocument: "required",
    fullDocumentBeforeChange: "required"
  }
)

结果验证

• 执行一个会导致变更事件 >16MB 的操作（如更新一个包含超大数组的文档）。

• 观察返回的事件流会被拆分成多个连续的fragment事件，最终以一个fragment事件结束。

降低前镜像存储开销

默认Pre-image随oplog过期。可设置更短的过期时间以节省空间：

db.adminCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: { expireAfterSeconds: 100 } // 单位：秒
    }
  }
})

1. 为什么 Change Stream 的慢日志总是显示 COLLSCAN？

这是正常且符合预期的行为，无需优化。Change Stream的cursor最终会建立在local.oplog.rs上（只有这一个地方记录了实例所有的修改），而该表没有索引。所以固定是COLLSCAN，无法避免也没有优化空间。

如果您在寻找待优化查询语句时使用了COLLSCAN作为日志过滤的关键字，推荐也额外过滤掉$changeStream关键字。

推荐在遇到changeStream性能问题（比如下游消费拉取存在延迟上涨）时才需要关注Change Stream相关的慢日志。

2. 为什么分片实例上对于非分片表建立监听，也能在除了primary shard之外的其他分片上观察到changeStream cursor？

为了应对可能存在的分片（shardCollection）操作。您可以随时对一个已经建立监听的非分片表执行shardCollection操作来将其数据分布到所有分片上（movePrimary类似）。提前在其他分片上建立Change Stream cursors就是为了应对这种情况。通常这些分片上的cursor并不会返回任何变更事件，性能开销也比较小。

与之类似的，为了应对可能存在的addShard/removeShard操作，分片实例上建立监听时，mongos也会到Config Server上去建立相关的cursor。

3. 为什么建立Change Stream cursor时指定了readPreference:secondary，但这个cursor的慢日志却出现在primary中？

游标（cursor）创建后会“固定”在某个节点上，不会因节点角色变化而自动迁移。因此一旦Change Stream的cursor建立之后，就固定在某个mongod节点（主或从节点）上并持续通过getMore消费。当发生过主从切换、变配、迁移或其他可能导致主从切换的事件时，原本在secondary上的cursor可能会转移到primary上。

如果不期望这部分负载持续在primary节点上消耗资源，则可以通过killCursors的方式来清理相关的changeStream cursor，下游消费逻辑处理会根据resumeToken并结合readPreference将cursor恢复到secondary节点上去，此操作不会导致变更事件的丢失或中断。

db.runCommand(
   {
     killCursors: <collection>,
     cursors: [ <cursor id1>, ... ], comment: <any>
   }
)
db.getSiblingDB("<testDB>").runCommand( { killCursors: "<testColl>", cursors: [NumberLong("2452840976689696187") ] } ) 

4. 为什么在mongos上看到大量关于Change Stream的长达1000ms的慢日志？

2020-08-26T04:34:45.045+0000 I COMMAND [conn21283] command altconfig-b2b-perf.oplog command: getMore \{ getMore: 3513599116181216748, collection: "oplog", $db: "altconfig-b2b-perf", $clusterTime: { clusterTime: Timestamp(1598416483, 1), signature: { hash: BinData(0, EC3841EB1FB7A34F897688BB5983E32E2ADF6763), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } originatingCommand: \{ find: "oplog", filter: { timestamp: { $gte: new Date(1597895390009) } }, tailable: true, awaitData: true, $db: "altconfig-b2b-perf", $clusterTime: \{ clusterTime: Timestamp(1597895487, 1), signature: { hash: BinData(0, AAAE33D5688F935C70469CBDB8EB1F6882749A40), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } nShards:1 cursorid:3513599116181216748 numYields:0 nreturned:0 reslen:237 protocol:op_msg 1000ms

这是旧版本 MongoDB （6.0 之前） 的正常现象，它代表的是等待超时而非性能瓶颈。在分片实例架构中，mongos到分片建立的Change Stream cursors会指定tailable:true、awaitData:true以及maxTimeMS:1000来尽可能在1000ms内返回更多的变更事件。

在6.0之前的大版本中，mongos上会记录这些长达1000ms的慢日志，存在误导客户的可能性。官方已经对相关行为进行了优化，细节可参考SERVER-50559。

5. 开启前镜像（Pre-image）后数据存储在哪里？

前镜像存储在config.system.preimages这个系统集合中，存在于每个 mongod 数据节点上（不是 Config Server）：

• 副本集：存在于各数据节点（primary + secondary 都有，随 oplog 复制）。

• 分片集群：存在于每个 shard 的 mongod 上，各 shard 独立存储自己处理的文档的前镜像。

后镜像不额外存储，会从 Change Stream event 中实时构造返回。

6. 开启前镜像（Pre-image）后会有什么开销？

• 存储开销：config.system.preimages集合的大小会持续增长，如果不配置过期策略或 oplog 窗口很大，可能占用大量磁盘空间。

• 写入开销：每次update/replace/delete操作都会额外写一份完整的旧文档到config.system.preimages，相当于写 I/O 翻倍（对大文档尤其明显）。

• WT cache 压力：config.system.preimages集合的数据也会进入 WiredTiger cache，挤占正常业务数据的缓存空间。

• oplog 记录：因为需要复制到副本集的从节点，因此前镜像文档的写入也会产生 oplog 条目。

• 16MB 限制：文档大小比较大时，包含前镜像的变更事件可能会超过 16MB 限制而导致变更流报错。6.0 版本由于还不支持$changeStreamSplitLargeEvent阶段，数据库无法自行处理，需要业务侧介入。