db.collection.watch()

定义

db.collection.watch( pipeline, options )

重要

mongosh 方法

本页面提供 mongosh 方法的相关信息。这不是数据库命令或特定语言驱动程序（例如 Node.js）的相关文档。

有关数据库命令，请参阅附带 $changeStream 聚合阶段的 aggregate 命令。

如需了解 MongoDB API 驱动程序，请参阅特定语言的 MongoDB 驱动程序文档。

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

打开针对某一集合的变更流游标。

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

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

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

返回：	只要与 MongoDB 部署的连接保持打开且集合存在，游标便会保持打开状态。请参阅变更事件，获取变更事件文档的示例。

提示

db.watch() 和 Mongo.watch()

兼容性

此方法可用于以下环境中托管的部署：

• MongoDB Atlas：用于云中 MongoDB 部署的完全托管服务

• MongoDB Enterprise：基于订阅、自我管理的 MongoDB 版本

• MongoDB Community：源代码可用、免费使用且可自行管理的 MongoDB 版本

可用性

行为

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

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

  • 游标已明确关闭。

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

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

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

访问控制

使用访问控制来运行时，用户必须对集合资源具有 find 和 changeStream 特权动作。换言之，用户必须具有能授予以下特权的角色：

{ resource: { db: <dbname>, collection: <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()
  }
}

示例

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

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

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

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

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

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

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 }
)

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