db.collection.watch()

Deployment部署

db.collection.watch() is available for replica set and sharded cluster deployments :可用于副本集和分片集群部署：

• For a replica set, you can issue db.collection.watch() on any data-bearing member.对于副本集，可以对任何承载数据的成员发出db.collection.watch()。
• For a sharded cluster, you must issue db.collection.watch() on a mongos instance.对于分片集群，必须在mongos实例上发出db.collection.watch()。

Storage Engine存储引擎

You can only use db.collection.watch() with the Wired Tiger storage engine.您只能将db.collection.watch()与Wired Tiger存储引擎一起使用。

Read Concern majority Support读取关注majority支持

Starting in MongoDB 4.2, change streams are available regardless of the "majority" read concern support; that is, read concern majority support can be either enabled (default) or disabled to use change streams.从MongoDB 4.2开始，无论"majority"读取关注支持如何，都可以使用更改流；也就是说，为了使用变更流，可以启用（默认）或禁用读取关注majority支持。

In MongoDB 4.0 and earlier, change streams are available only if "majority" read concern support is enabled (default).在MongoDB 4.0及更早版本中，只有启用了"majority"读取关注支持（默认），更改流才可用。

Resumability可恢复性

Unlike the MongoDB Drivers, mongosh does not automatically attempt to resume a change stream cursor after an error. 与MongoDB驱动程序不同，mongosh不会在出现错误后自动尝试恢复更改流游标。The MongoDB drivers make one attempt to automatically resume a change stream cursor after certain errors.MongoDB驱动程序尝试在出现某些错误后自动恢复更改流游标。

db.collection.watch() uses information stored in the oplog to produce the change event description and generate a resume token associated to that operation. 使用oplog中存储的信息来生成更改事件描述，并生成与该操作相关联的恢复令牌。If the operation identified by the resume token passed to the resumeAfter or startAfter option has already dropped off the oplog, db.collection.watch() cannot resume the change stream.如果传递给resumeAfter或startAfter选项的resume令牌标识的操作已经从操作日志中删除，则db.collection.watch()无法恢复更改流。

See Resume a Change Stream for more information on resuming a change stream.有关恢复更改流的详细信息，请参阅恢复更改流。

Full Document Lookup of Update Operations更新操作的完整文档查找

By default, the change stream cursor returns specific field changes/deltas for update operations. 默认情况下，更改流游标返回用于更新操作的特定字段更改/增量。You can also configure the change stream to look up and return the current majority-committed version of the changed document. 您还可以配置更改流以查找并返回更改文档的当前多数提交版本。Depending on other write operations that may have occurred between the update and the lookup, the returned document may differ significantly from the document at the time of the update.根据更新和查找之间可能发生的其他写入操作，返回的文档可能与更新时的文档有很大不同。

Depending on the number of changes applied during the update operation and the size of the full document, there is a risk that the size of the change event document for an update operation is greater than the 16MB BSON document limit. 根据更新操作期间应用的更改数量和完整文档的大小，更新操作的更改事件文档的大小可能大于16MB BSON文档限制。If this occurs, the server closes the change stream cursor and returns an error.如果发生这种情况，服务器将关闭更改流游标并返回一个错误。

Open a Change Stream打开变更流

The following operation opens a change stream cursor against the data.sensors collection:以下操作打开data.sensors集合上的更改流游标：

watchCursor = db.getSiblingDB("data").sensors.watch()

Iterate the cursor to check for new events. 循环游标以检查新事件。Use the cursor.isClosed() method with the cursor.tryNext() method to ensure the loop only exits if the change stream cursor is closed and there are no objects remaining in the latest batch:将cursor.isClosed()方法与cursor.tryNext()方法结合使用，以确保只有在更改流游标关闭并且最新批中没有剩余对象时，循环才会退出：

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

For complete documentation on change stream output, see Change Events.有关变更流输出的完整文档，请参阅变更事件。

Change Stream with Full Document Update Lookup使用完整文档更新查找更改流

Set the fullDocument option to "updateLookup" to direct the change stream cursor to lookup the most current majority-committed version of the document associated to an update change stream event.将fullDocument选项设置为"updateLookup"，以指示变更流游标查找与更新变更流事件相关联的文档的最新多数提交版本。

The following operation opens a change stream cursor against the data.sensors collection using the fullDocument : "updateLookup" option.以下操作使用fullDocument : "updateLookup"选项打开data.sensors集合的更改流游标。

watchCursor = db.getSiblingDB("data").sensors.watch(
   [],
   { fullDocument : "updateLookup" }
)

Iterate the cursor to check for new events. 循环游标以检查新事件。Use the cursor.isClosed() method with the cursor.tryNext() method to ensure the loop only exits if the change stream cursor is closed and there are no objects remaining in the latest batch:将cursor.isClosed()方法与cursor.tryNext()方法结合使用，以确保只有在更改流游标关闭并且最新批中没有剩余对象时，循环才会退出：

while (!watchCursor.isClosed()) {
  let next = watchCursor.tryNext()
  while (next !== null) {
    printjson(next);
    next = watchCursor.tryNext()
  }
}

For any update operation, the change event returns the result of the document lookup in the fullDocument field.对于任何更新操作，更改事件都会在fullDocument字段中返回文档查找的结果。

For an example of the full document update output, see change stream update event.有关完整文档更新输出的示例，请参阅变更流更新事件。

For complete documentation on change stream output, see Change Events.有关变更流输出的完整文档，请参阅更改事件。

Change Streams with Document Pre- and Post-Images使用文档前后映像更改流

Starting in MongoDB 6.0, you can use change stream events to output the version of a document before and after changes (the document pre- and post-images):从MongoDB 6.0开始，您可以使用更改流事件来输出更改前后文档的版本（文档前后映像）：

• The pre-image is the document before it was replaced, updated, or deleted. 前映像是在文档被替换、更新或删除之前的文档。There is no pre-image for an inserted document.插入的文档没有前映像。
• The post-image is the document after it was inserted, replaced, or updated. 后映像是插入、替换或更新后的文档。There is no post-image for a deleted document.已删除的文档没有后映像。
• Enable changeStreamPreAndPostImages for a collection using db.createCollection(), create, or collMod.使用db.createCollection()、create或collMod为集合启用changeStreamPreAndPostImages。

Pre- and post-images are not available for a change stream event if the images were:如果映像为以下情况，则前映像和后映像对变更流事件不可用：

• Not enabled on the collection at the time of a document update or delete operation.文档更新或删除操作时未在集合上启用。
• Removed after the pre- and post-image retention time set in expireAfterSeconds.在expireAfterSeconds中设置的前映像和后映像保留时间之后删除。
  • The following example sets expireAfterSeconds to 100 seconds:以下示例将expireAfterSeconds设置为100秒：

    use admin
    db.runCommand( {
       setClusterParameter:
          { changeStreamOptions: { preAndPostImages: { expireAfterSeconds: 100 } } }
    } )

  • The following example returns the current changeStreamOptions settings, including expireAfterSeconds:以下示例返回当前changeStreamOptions设置，包括expireAfterSeconds：

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

  • Setting expireAfterSeconds to off uses the default retention policy: pre- and post-images are retained until the corresponding change stream events are removed from the oplog.将expireAfterSeconds设置为off使用默认的保留策略：保留前映像和后映像，直到从oplog中删除相应的更改流事件。
  • If a change stream event is removed from the oplog, then the corresponding pre- and post-images are also deleted regardless of the expireAfterSeconds pre- and post-image retention time.如果更改流事件从操作日志中删除，则相应的前映像和后映像也将被删除，而不管映像前和后映像的保留时间是expireAfterSeconds。

Additional considerations:其他注意事项：

• Enabling pre- and post-images consumes storage space and adds processing time. 启用前映和后映像会消耗存储空间并增加处理时间。Only enable pre- and post-images if you need them.只有在需要时才启用前映像和后映像。
• Limit the change stream event size to less than 16 megabytes. To limit the event size, you can:将更改流事件大小限制为小于16兆字节。要限制事件大小，您可以：
  • Limit the document size to 8 megabytes. 将文档大小限制为8兆字节。You can request pre- and post-images simultaneously in the change stream output if other change stream event fields like updateDescription are not large.如果其他变更流事件字段（如updateDescription）不大，则可以在变更流输出中同时请求前映像和后映像。
  • Request only post-images in the change stream output for documents up to 16 megabytes if other change stream event fields like updateDescription are not large.如果其他更改流事件字段（如updateDescription）不太大，则请求在更改流输出中为高达16兆字节的文档后映像。
  • Request only pre-images in the change stream output for documents up to 16 megabytes if:如果出现以下情况，则仅在更改流输出中请求高达16兆字节的文档的前映像：
    • document updates affect only a small fraction of the document structure or content, and文档更新只影响文档结构或内容的一小部分，并且
    • do not cause a replace change event. A replace event always includes the post-image.不要引起replace更改事件。replace事件始终包括后映像。
• To request a pre-image, you set fullDocumentBeforeChange to required or whenAvailable in db.collection.watch(). To request a post-image, you set fullDocument using the same method.若要请求前映像，请在db.collection.watch()中将fullDocumentBeforeChange设置为required或whenAvailable。若要请求后映像，请使用相同的方法设置fullDocument。
• Pre-images are written to the config.system.preimages collection.前映像将写入config.system.preimages集合。
  • The config.system.preimages collection may become large. config.system.preimages集合可能会变大。To limit the collection size, you can set expireAfterSeconds time for the pre-images as shown earlier.要限制集合大小，您可以为前映像设置expireAfterSeconds时间，如前所示。
  • Pre-images are removed asynchronously by a background process.前映像通过后台进程异步移除。

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

db.temperatureSensor.insertMany( [
   { "_id" : 0, "reading" : 26.1 },
   { "_id" : 1, "reading" : 25.9 },
   { "_id" : 2, "reading" : 24.3 },
   { "_id" : 3, "reading" : 22.4 },
   { "_id" : 4, "reading" : 24.6 }
] )

watchCursorFullDocumentBeforeChange = db.temperatureSensor.watch(
   [],
   { fullDocumentBeforeChange: "whenAvailable" }
)

while ( !watchCursorFullDocumentBeforeChange.isClosed() ) {
   if ( watchCursorFullDocumentBeforeChange.hasNext() ) {
      printjson( watchCursorFullDocumentBeforeChange.next() );
   }
}

db.temperatureSensor.updateOne(
   { _id: 2 },
   { $set: { reading: 22.1 } }
)

{
   "_id" : {
      "_data" : "82624B21...",
      "_typeBits" : BinData(0,"QA==")
   },
   "operationType" : "update",
   "clusterTime" : Timestamp(1649090957, 1),
   "ns" : {
      "db" : "test",
      "coll" : "temperatureSensor"
   },
   "documentKey" : {
      "_id" : 2
   },
   "updateDescription" : {
      "updatedFields" : {
         "reading" : 22.1
      },
      "removedFields" : [ ],
      "truncatedArrays" : [ ]
   },
   "fullDocumentBeforeChange" : {
      "_id" : 2,
      "reading" : 24.3
   }
}

watchCursorFullDocument = db.temperatureSensor.watch(
   [],
   { fullDocument: "whenAvailable" }
)

while ( !watchCursorFullDocument.isClosed() ) {
   if ( watchCursorFullDocument.hasNext() ) {
      printjson( watchCursorFullDocument.next() );
   }
}

db.temperatureSensor.updateOne(
   { _id: 1 },
   { $set: { reading: 29.5 } }
)

{
   "_id" : {
      "_data" : "8262474D...",
      "_typeBits" : BinData(0,"QA==")
   },
   "operationType" : "update",
   "clusterTime" : Timestamp(1648840090, 1),
   "fullDocument" : {
      "_id" : 1,
      "reading" : 29.5
   },
   "ns" : {
      "db" : "test",
      "coll" : "temperatureSensor"
   },
   "documentKey" : {
      "_id" : 1
   },
   "updateDescription" : {
      "updatedFields" : {
         "reading" : 29.5
      },
      "removedFields" : [ ],
      "truncatedArrays" : [ ]
   }
}

Change Stream with Aggregation Pipeline Filter使用聚合管道筛选器更改流

The following operation opens a change stream cursor against the data.sensors collection using an aggregation pipeline to filter only insert events:以下操作使用聚合管道打开data.sensors集合的更改流游标，以仅筛选insert事件：

watchCursor = db.getSiblingDB("data").sensors.watch(
   [
      { $match : {"operationType" : "insert" } }
   ]
)

Iterate the cursor to check for new events. Use the cursor.isClosed() method with the cursor.hasNext() method to ensure the loop only exits if the change stream cursor is closed and there are no objects remaining in the latest batch:循环游标以检查新事件。将cursor.isClosed()方法与cursor.hasNext()方法结合使用，以确保只有在更改流游标关闭且最新批中没有剩余对象时，循环才会退出：

while (!watchCursor.isClosed()){
   if (watchCursor.hasNext()){
      printjson(watchCursor.next());
   }
}

The change stream cursor only returns change events where the operationType is insert. 更改流游标只返回operationType为insert的更改事件。For complete documentation on change stream output, see Change Events.有关变更流输出的完整文档，请参阅变更事件。

Resuming a Change Stream恢复更改流

Every document returned by a change stream cursor includes a resume token as the _id field. 变更流游标返回的每个文档都包含一个resume令牌作为_id字段。To resume a change stream, pass the entire _id document of the change event you want to resume from to either the resumeAfter or startAfter option of watch().要恢复更改流，请将要恢复的更改事件的整个_id文档传递给watch()的resumeAfter或startAfter选项。

The following operation resumes a change stream cursor against the data.sensors collection using a resume token. This assumes that the operation that generated the resume token has not rolled off the cluster's oplog.以下操作使用恢复令牌恢复data.sensors集合上的更改流游标。这假设生成恢复令牌的操作尚未从集群的操作日志中滚出。

let watchCursor = db.getSiblingDB("data").sensors.watch();
let firstChange;

while (!watchCursor.isClosed()) {
   if (watchCursor.hasNext()) {
     firstChange = watchCursor.next();
     break;
   }
}

watchCursor.close();

let resumeToken = firstChange._id;

resumedWatchCursor = db.getSiblingDB("data").sensors.watch(
[],
   { resumeAfter : resumeToken }
)

Iterate the cursor to check for new events. 循环游标以检查新事件。Use the cursor.isClosed() method with the cursor.hasNext() method to ensure the loop only exits if the change stream cursor is closed and there are no objects remaining in the latest batch:将cursor.isClosed()方法与cursor.hasNext()方法结合使用，以确保只有在更改流游标关闭且最新批中没有剩余对象时，循环才会退出：

while (!resumedWatchCursor.isClosed()){
   if (resumedWatchCursor.hasNext()){
      print(resumedWatchCursor.next());
   }
}

See Resume a Change Stream for complete documentation on resuming a change stream.有关恢复更改流的完整文档，请参阅恢复更改流。