MongoDB 3.6引入change streams，如何在Mongo Shell中创建该流？

当然可以！MongoDB 3.6及以上版本的Mongo Shell完全支持Change Streams功能，而且用起来相当直观，我来给你一步步演示怎么操作以及一些实用技巧。

前提条件 #

• 你的MongoDB部署必须是副本集或者分片集群：Change Streams依赖于Oplog来捕获变更，单节点独立实例无法提供这个能力。
• 确保你的Mongo Shell版本与服务器版本匹配（至少为3.6）。

基础使用示例 #

监听单个集合的所有变更 #

打开Mongo Shell并连接到目标部署后，执行以下命令：

// 切换到目标数据库
use your_target_db

// 针对目标集合创建Change Stream
const changeStream = db.your_target_collection.watch()

// 持续读取并打印变更事件
while (changeStream.hasNext()) {
  printjson(changeStream.next())
}

执行后，Shell会持续监听该集合的所有变更操作（插入、更新、删除、替换等），每产生一个变更就会输出对应的事件文档。

过滤特定类型的变更 #

如果你只关心某一类操作（比如只监听插入事件），可以通过传入聚合管道来过滤：

// 定义过滤管道：只匹配插入操作
const filterPipeline = [
  { $match: { operationType: 'insert' } }
]

// 创建带过滤条件的Change Stream
const changeStream = db.your_target_collection.watch(filterPipeline)

while (changeStream.hasNext()) {
  printjson(changeStream.next())
}

支持的operationType枚举值包括：insert、update、delete、replace、drop、rename等。

监听整个数据库的变更 #

除了集合级别，你还可以监听整个数据库下所有集合的变更：

const dbLevelStream = db.watch()

while (dbLevelStream.hasNext()) {
  printjson(dbLevelStream.next())
}

实用技巧与注意事项 #

• 停止监听：可以直接按Ctrl+C中断Shell的循环，或者调用changeStream.close()方法主动关闭流。
• 获取更新后的完整文档：默认情况下，update类型的事件不会返回完整的更新后文档，你可以在创建流时指定fullDocument: 'updateLookup'参数来获取：

  const changeStream = db.your_target_collection.watch([], { fullDocument: 'updateLookup' })
  

• 断点续传：如果需要在断开重连后从上次中断的位置继续监听，可以利用resumeToken：

  let lastResumeToken = null;
  const changeStream = db.your_target_collection.watch()
  
  while (changeStream.hasNext()) {
    const event = changeStream.next()
    printjson(event)
    // 记录当前的resumeToken
    lastResumeToken = event._id
  }
  
  // 后续重连时，通过resumeToken恢复监听
  const resumedStream = db.your_target_collection.watch([], { resumeAfter: lastResumeToken })
  

内容的提问来源于stack exchange，提问作者Paras Diwan