Mongo.watch()

Definition定义

Mongo.watch( pipeline, options )

For replica sets and sharded clusters only仅适用于复制集和分片集群

Opens a change stream cursor for a replica set or a sharded cluster to report on all its non-system collections across its databases, with the exception of the admin, local, and the config databases.打开副本集或分片集群的更改流游标，以报告其数据库中的所有非系统集合，admin数据库、local数据库和config数据库除外。

Parameter参数	Type类型	Description描述
pipeline	array	Optional.可选的。An Aggregation Pipeline consisting of one or more of the following aggregation stages: 由以下一个或多个聚合阶段组成的聚合管道： $addFields $match $project $replaceRoot $replaceWith (Available starting in MongoDB 4.2)（从MongoDB 4.2开始提供） $redact $set (Available starting in MongoDB 4.2)（从MongoDB 4.2开始提供） $unset (Available starting in MongoDB 4.2)（从MongoDB 4.2开始提供） Specify a pipeline to filter/modify the change events output.指定用于筛选/修改更改事件输出的管道。 Starting in MongoDB 4.2, change streams will throw an exception if the change stream aggregation pipeline modifies an event's _id field. 从MongoDB 4.2开始，如果变更流聚合管道修改了事件的_id字段，则变更流将抛出异常。
options	document	Optional.可选的。Additional options that modify the behavior of Mongo.watch().修改Mongo.watch()行为的其他选项。

The options document can contain the following fields and values:options文档可以包含以下字段和值：

Field字段	Type类型	Description描述
resumeAfter	document	Optional.可选的。Directs Mongo.watch() to attempt resuming notifications starting after the operation specified in the resume token.指示Mongo.watch()尝试在恢复令牌中指定的操作之后开始恢复通知。 Each change stream event document includes a resume token as the _id field. 每个变更流事件文档都包括一个恢复令牌作为_id字段。Pass the entire _id field of the change event document that represents the operation you want to resume after.传递更改事件文档的整个_id字段，该字段表示要在之后恢复的操作。 resumeAfter is mutually exclusive with startAfter and startAtOperationTime. 与startAfter和startAtOperationTime互斥。
startAfter	document	Optional.可选的。Directs Mongo.watch() to attempt starting a new change stream after the operation specified in the resume token. 指示Mongo.watch()在恢复令牌中指定的操作之后尝试启动新的更改流。Allows notifications to resume after an invalidate event.允许在发生无效事件后恢复通知。 Each change stream event document includes a resume token as the _id field. 每个变更流事件文档都包括一个恢复令牌作为_id字段。Pass the entire _id field of the change event document that represents the operation you want to resume after.传递更改事件文档的整个_id字段，该字段表示要在之后恢复的操作。 startAfter is mutually exclusive with resumeAfter and startAtOperationTime. startAfter与resumeAfter和startAtOperationTime互斥。 New in version 4.2. 4.2版新增。
fullDocument	string	Optional.可选的。By default, Mongo.watch() returns the delta of those fields modified by an update operation, instead of the entire updated document.默认情况下，Mongo.watch()返回由更新操作修改的字段的delta，而不是整个更新的文档。 Set fullDocument to "updateLookup" to direct Mongo.watch() to look up the most current majority-committed version of the updated document. 将fullDocument设置为"updateLookup"以指示Mongo.watch()查找更新文档的最新多数提交版本。Mongo.watch() returns a fullDocument field with the document lookup in addition to the updateDescription delta. 除了updateDescription delta之外，还返回一个带有文档查找的fullDocument字段。
batchSize	int	Optional.可选的。Specifies the maximum number of change events to return in each batch of the response from the MongoDB cluster.指定在MongoDB集群的每一批响应中返回的最大更改事件数。 Has the same functionality as cursor.batchSize(). 具有与cursor.batchSize()相同的功能。
maxAwaitTimeMS	int	Optional.可选的。The maximum amount of time in milliseconds the server waits for new data changes to report to the change stream cursor before returning an empty batch.服务器在返回空批之前等待新数据更改报告给更改流游标的最长时间（以毫秒为单位）。 Defaults to 1000 milliseconds. 默认值为1000毫秒。
collation	document	Optional.可选的。Pass a collation document to specify collation for the change stream cursor.传递排序规则文档以指定更改流游标的排序规则。 If omitted, defaults to simple binary comparison. 如果省略，则默认为simple二进制比较。
startAtOperationTime	Timestamp	Optional.可选的。The starting point for the change stream. 变更流的起点。If the specified starting point is in the past, it must be in the time range of the oplog. 如果指定的起点在过去，则它必须在oplog的时间范围内。To check the time range of the oplog, see rs.printReplicationInfo().要检查oplog的时间范围，请参阅rs.printReplicationInfo()。 startAtOperationTime is mutually exclusive with resumeAfter and startAfter. startAtOperationTime与resumeAfter和startAfter互斥。

Returns:返回值：	A cursor over the change event documents. See Change Events for examples of change event documents.将游标悬停在更改事件文档上。有关更改事件文档的示例，请参阅更改事件。

Tip

See also: 另请参阅：

db.collection.watch() and 和db.watch()

Availability可用性

Behavior行为

• Mongo.watch() only notifies on data changes that have persisted to a majority of data-bearing members.仅通知已持久存在于大多数数据承载成员的数据更改。
• The change stream cursor remains open until one of the following occurs:更改流游标保持打开状态，直到出现以下情况之一：
  • The cursor is explicitly closed.游标明确关闭。
  • An invalidate event occurs; for example, a collection drop or rename.发生失效事件；例如集合删除或重命名。
  • The connection to the MongoDB deployment is closed.与MongoDB部署的连接已关闭。
  • If the deployment is a sharded cluster, a shard removal may cause an open change stream cursor to close, and the closed change stream cursor may not be fully resumable.如果部署是分片集群，则分片移除可能会导致打开的变更流游标关闭，并且关闭的变更流游标可能无法完全恢复。

snippet install resumetoken
decodeResumeToken('<RESUME TOKEN>')

npx mongodb-resumetoken-decoder <RESUME TOKEN>

Access Control访问控制

When running with access control, the user must have the find and changeStream privilege actions on all non-systems collections across all databases. 在使用访问控制运行时，用户必须对所有数据库中的所有非系统集合具有find和changeStream权限操作。That is, a user must have a role that grants the following privilege:也就是说，用户必须具有授予以下权限的角色：

{ resource: { db: "", collection: "" }, actions: [ "find", "changeStream" ] }

The built-in read role provides the appropriate privileges.内置的read角色提供适当的权限。

Cursor Iteration游标迭代

MongoDB provides multiple ways to iterate on a cursor.MongoDB提供了多种对游标进行迭代的方法。

The cursor.hasNext() method blocks and waits for the next event. To monitor the watchCursor cursor and iterate over the events, use hasNext() like this:cursor.hasNext()方法阻塞并等待下一个事件。要监视watchCursor游标并迭代事件，请使用hasNext()，如下所示：

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

The cursor.tryNext() method is non-blocking. cursor.tryNext()方法是非阻塞的。To monitor the watchCursor cursor and iterate over the events, use tryNext() like this:要监视watchCursor游标并迭代事件，请使用tryNext()，如下所示：

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

Example实例

The following operation in mongosh opens a change stream cursor on a replica set. mongosh中的以下操作将打开复制集上的更改流游标。The returned cursor reports on data changes to all the non-system collections across all databases except for the admin, local, and the config databases.返回的游标报告除admin、local和config数据库之外的所有数据库中所有非system集合的数据更改。

watchCursor = db.getMongo().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.有关变更流输出的完整文档，请参阅变更事件。