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, the mongo shell does not automatically attempt to resume a change stream cursor after an error. 与MongoDB驱动程序不同，mongo shell不会在出现错误后自动尝试恢复更改流游标。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令牌所标识的操作已从oplog中删除，则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.isExhausted() 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.isExhausted()方法确保循环仅在更改流光标关闭且最新批处理中没有剩余对象时退出：

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

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.isExhausted() 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.isExhausted()方法确保循环仅在更改流光标关闭且最新批处理中没有剩余对象时退出：

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

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 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.isExhausted() 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.isExhausted()方法确保循环仅在更改流光标关闭且最新批处理中没有剩余对象时退出：

while (!watchCursor.isExhausted()){
   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. 下面的操作使用恢复令牌针对data.sensors集合恢复更改流光标。This assumes that the operation that generated the resume token has not rolled off the cluster’s oplog.这假设生成恢复令牌的操作没有从集群的oplog中滚出来。

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

while (!watchCursor.isExhausted()) {
   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.isExhausted() 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.isExhausted()方法确保循环仅在更改流光标关闭且最新批处理中没有剩余对象时退出：

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

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