Mongo.watch()

定义

Mongo.watch( pipeline, options )

仅适用于副本集和分片集群

为副本集或分片集群打开变更流游标，可报告其数据库中的所有非 system 集合，admin、local 和 config 数据库除外。

范围	类型	说明
pipeline	阵列	可选。Aggregation Pipeline 由以下一个或多个聚合阶段组成： $addFields $match $project $replaceRoot $replaceWith （从 MongoDB 4.2 开始可用） $redact $set （从 MongoDB 4.2 开始可用） $unset （从 MongoDB 4.2 开始可用） 指定用于筛选/修改变更事件输出的管道。 从 MongoDB 4.2 开始，如果变更流聚合管道修改了事件的 _id 字段，则变更流会引发异常。
options	文档	可选。用于修改 Mongo.watch()行为的其他选项。

options 文档可包含以下字段和值：

字段	类型	说明
resumeAfter	文档	可选。指示Mongo.watch()尝试在恢复令牌中指定的操作之后开始恢复通知。 每份变更流事件文档都包含一个恢复令牌作为 _id 字段。传递变更事件文档的整个 _id 字段，代表之后想要恢复的操作。 resumeAfter 与 startAfter 和 startAtOperationTime 互斥。
startAfter	文档	可选。指示Mongo.watch()在恢复令牌中指定的操作后尝试启动新的变更流。允许在无效事件后恢复通知。 每份变更流事件文档都包含一个恢复令牌作为 _id 字段。传递变更事件文档的整个 _id 字段，代表之后想要恢复的操作。 startAfter 与 resumeAfter 和 startAtOperationTime 互斥。 4.2 版本中的新增功能。
fullDocument	字符串	可选。默认情况下， Mongo.watch()返回通过更新操作修改的字段的增量，而不是整个更新后的文档。 将fullDocument设置为"updateLookup" ，以指示Mongo.watch()查找更新文档的最新多数提交版本。 Mongo.watch()除了updateDescription增量之外，还会返回一个包含文档查找的fullDocument字段。
batchSize	int	可选。指定从 MongoDB 集群响应的每个批次中返回的变更事件的最大数量。 功能与 cursor.batchSize() 相同。
maxAwaitTimeMS	int	可选。在返回空批次之前，服务器等待新数据更改以报告给变更流游标的最长时间（以毫秒为单位）。 默认值为 1000 毫秒。
collation	文档	可选。传递排序规则文档以指定变更流游标的排序规则。 如果省略，则默认为 simple 二进制比较。
startAtOperationTime	时间戳	可选。变更流的起点。如果指定的起点在过去，它必须在 oplog 的时间范围内。要检查 oplog 的时间范围，请参阅 rs.printReplicationInfo()。 startAtOperationTime 与 resumeAfter 和 startAfter 互斥。

返回：	游标位于更改事件文档上。有关更改事件文档的示例，请参阅更改事件。

提示

另请参阅：

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

可用性

行为

• Mongo.watch() 仅通知持续到大多数数据承载节点的数据更改。

• 变更流游标保持打开状态，直到出现以下任一情况：

  • 游标已明确关闭。

  • 发生失效事件；例如删除或重命名集合。

  • 与 MongoDB 部署之间的连接关闭或超时。有关更多信息，请参阅游标行为。

  • 如果部署是分片集群，则分片删除可能会导致打开的变更流游标关闭，而关闭的变更流游标可能无法完全恢复。

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

npx mongodb-resumetoken-decoder <RESUME TOKEN>

访问控制

使用访问控制运行时，用户必须对所有数据库中的所有非系统集合具有 find 和 changeStream 权限操作。换言之，用户必须具有能授予以下特权的角色：

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

内置 read 角色提供了相应的特权。

游标迭代

MongoDB 提供多种在游标上迭代的方法。

cursor.hasNext() 方法会阻塞并等待下一个事件。要监控 watchCursor 游标并迭代事件，请使用 hasNext()，如下所示：

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

该 cursor.tryNext() 方法为非阻塞式。要监控 watchCursor 游标并迭代事件，请使用 tryNext()，如下所示：

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

例子

mongosh中的以下操作在副本集上打开变更流游标。返回的游标报告所有数据库中所有非system集合的数据更改， admin 、 local和config数据库除外。

watchCursor = db.getMongo().watch()

迭代游标以检查是否存在新事件。使用 cursor.isClosed() 方法和 cursor.tryNext() 方法，确保循环仅在变更流游标已关闭且最新批次中没有剩余对象时才退出：

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

如需变更流输出的完整文档，请参阅变更事件。