使用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相关信息。

   

开启前镜像（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" } )

   开启成功后输出如下信息：

2. 检查集合配置：

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

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

   

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

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

   

更多信息参考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 字段（变更后文档）。

处理超大变更事件 (>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 } // 单位：秒
    }
  }
})