db.collection.watch（）

定义¶

db.collection.watch（管道，选项）¶

mongo 壳法

此页面记录了mongoshell方法，并且 没有引用MongoDB Node.js驱动程序（或任何其他驱动程序）方法。对于相应的MongoDB驱动程序API，请改为参考您特定的 MongoDB驱动程序文档。

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

在集合上打开更改流游标。

参数	类型	描述
pipeline	数组	由以下一个或多个聚合阶段组成的聚合管道： $addFields $match $project $replaceRoot $replaceWith （从MongoDB 4.2开始可用） $redact $set （从MongoDB 4.2开始可用） $unset （从MongoDB 4.2开始可用） 指定管道以过滤/修改变更事件输出。 从MongoDB 4.2开始，如果更改流聚合管道修改了事件的_id字段，则更改流将引发异常。
options	文献	可选的。修改行为​​的其他选项 watch()。 如果您未指定管道但正在传递 文档，则必须将空数组传递[]给pipeline参数options。

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

领域	类型	描述
resumeAfter	文献	可选的。指示watch尝试在恢复令牌中指定的操作之后开始恢复通知。 每个变更流事件文档都包括一个恢复令牌作为 _id字段。传递更改事件文档的整个 _id字段，该字段代表您要恢复的操作。 resumeAfter与startAfter和 互斥startAtOperationTime。
startAfter	文献	可选的。指示watch在恢复令牌中指定的操作之后尝试启动新的更改流。允许在事件无效后恢复通知。 每个变更流事件文档都包括一个恢复令牌作为 _id字段。传递更改事件文档的整个 _id字段，该字段代表您要恢复的操作。 startAfter与resumeAfter和 互斥startAtOperationTime。 4.2版中的新功能。
fullDocument	串	可选的。默认情况下，watch()返回通过更新操作修改的那些字段的增量，而不是整个更新的文档。 设置fullDocument为“ "updateLookup"直接” watch()以查找更新文档的最新多数批准版本。 watch()返回一个fullDocument字段，其中包含除了updateDescription增量之外的文档查找。
batchSize	整型	可选的。指定从MongoDB集群的每批响应中返回的更改事件的最大数量。 具有与相同的功能cursor.batchSize()。
maxAwaitTimeMS	整型	可选的。服务器等待新数据更改以报告更改流游标之前等待的最大时间（以毫秒为单位），然后返回空批处理。 默认为1000毫秒。
collation	文献	可选的。传递归类文档以为更改流游标 指定归类。 从MongoDB 4.2开始，simple如果省略，默认为二进制比较。在早期版本中，在单个集合上打开的更改流将继承集合的默认排序规则。
startAtOperationTime	时间戳记	可选的。变更流的起点。如果指定的起点是过去的时间，则必须在操作日志的时间范围内。要查看操作日志的时间范围，请参阅 rs.printReplicationInfo()。 startAtOperationTime与resumeAfter 和互斥startAfter。 版本4.0中的新功能。

返回值：	一个游标是保持被打开，以MongoDB的部署的连接保持打开状态，并收集存在。有关变更事件文档的示例，请参见变更事件。

也可以看看

db.watch() 和 Mongo.watch()

可用性¶

行为¶

• db.collection.watch() 仅通知已保留给大多数数据承载成员的数据更改。
• 该变换流光标保持打开状态，直到发生以下情况之一：
  • 游标已明确关闭。
  • 发生无效事件；例如，收藏夹放置或重命名。
  • 与MongoDB部署的连接已关闭。
  • 如果部署是分片群集，则分片删除可能会导致打开的更改流游标关闭，并且关闭的更改流游标可能无法完全恢复。

访问控制¶

使用访问控制运行时，用户必须对集合资源具有 find和changeStream特权操作。也就是说，用户必须具有授予以下特权的角色：

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

内置read角色提供适当的特权。

例子¶

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

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

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

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

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

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

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

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