更改流

本页内容

可用性

连接
监视集合、数据库或部署
更改流性能考虑因素
打开更改流
修改更改流输出
更新操作查找完整文档
恢复更改流
用例
访问控制
事件通知
排序规则
更改流和孤儿文档
具有文档前后图像的更改流

更改流允许应用程序在不进行手动跟踪的复杂性和风险的情况下访问实时数据更改。oplog。应用程序可以使用变更流来订阅单个集合、数据库或整个部署上的所有数据更改，并立即做出反应。由于变更流使用聚合框架，应用程序还可以过滤特定的更改或随意转换通知。

从 MongoDB 5.1 开始，变更流进行了优化，提供了更有效的资源利用率和一些聚合管道阶段的更快执行。

可用性

变更流适用于副本集和分片集群：

存储引擎。
副本集和分片集群必须使用 WiredTiger 存储引擎。变更流也可以用于使用 MongoDB 的静态加密功能的部署。
副本集协议版本。
副本集和分片集群必须使用副本集协议版本 1 (pv1)。
读取关注“多数”启用。
变更流无论是否支持 "majority" 读取关注；也就是说，读取关注 majority 支持可以是启用（默认）或禁用以使用变更流。

稳定的API支持

更改流包含在稳定API V1中。然而，showExpandedEvents选项不包括在稳定API V1中。

连接

更改流的连接可以使用带有+srv连接选项的DNS种子列表，或者在连接字符串中单独列出服务器。

如果驱动程序失去了对更改流的连接或连接中断，它将通过具有匹配的读取偏好的集群中的另一个节点尝试重新建立对更改流的连接。如果驱动程序找不到具有正确读取偏好的节点，它将抛出异常。

有关更多信息，请参阅连接字符串URI格式。

监视集合、数据库或部署

您可以对以下对象打开更改流

目标	描述
一个集合	您可以为单个集合打开更改流游标（除了`system`集合，或者任何位于`admin`、`local`和`config`数据库中的集合）。本页面的示例使用 MongoDB 驱动程序打开并操作单个集合的变更流游标。请参阅`mongosh`方法`db.collection.watch()`。
数据库	您可以为单个数据库（不包括`admin`、`local`和`config`数据库）打开变更流游标，以监视所有非系统集合的更改。有关 MongoDB 驱动程序方法，请参阅您的驱动程序文档。请参阅`mongosh`方法`db.watch()`。
部署	您可以为部署（无论是复制集还是分片集群）打开变更流游标，以监视除`admin`、`local`和`config`之外的所有数据库中的所有非系统集合的更改。有关 MongoDB 驱动程序方法，请参阅您的驱动程序文档。请参阅`mongosh`方法`Mongo.watch()`。

注意

变更流示例

本页面的示例使用 MongoDB 驱动程序说明了如何打开集合的变更流游标以及如何处理变更流游标。

变更流性能考虑因素

如果数据库上打开的活动变更流数量超过连接池大小，您可能会遇到通知延迟。每个变更流在等待下一个事件期间都会使用一个连接和一个getMore操作。为了避免任何延迟问题，您应确保池大小大于打开的变更流数量。有关详细信息，请参阅maxPoolSize设置。

分片集群考虑事项

当在分片集群上打开更改流时

无论更改流的目标是否为特定的分片键范围，mongos 都会在每个分片上创建单独的更改流。这种行为不受更改流目标是否为特定分片键范围的影响。
当 mongos 接收到更改流结果时，它会对这些结果进行排序和过滤。如果需要，mongos 还会执行 fullDocument 查找。

为了获得最佳性能，请限制在更改流中使用 $lookup 查询。

打开更改流

要打开更改流

对于副本集，您可以从任何承载数据的成员发出打开更改流操作。
对于分片集群，您必须从 mongos 发出打开更改流操作。

以下示例打开一个集合的更改流，并遍历游标以检索更改流文档。 [1]

➤使用页面上方右边的“选择您的语言”下拉菜单设置此页面上示例的语言。

以下C语言示例假设您已连接到MongoDB副本集并访问了数据库，并且包含一个inventory集合。

mongoc_collection_t *collection;
bson_t *pipeline = bson_new ();
bson_t opts = BSON_INITIALIZER;
mongoc_change_stream_t *stream;
const bson_t *change;
const bson_t *resume_token;
bson_error_t error;
collection = mongoc_database_get_collection (db, "inventory");
stream = mongoc_collection_watch (collection, pipeline, NULL /* opts */);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}
mongoc_change_stream_destroy (stream);

以下C#示例假设您已连接到MongoDB副本集并访问了数据库，并且包含一个inventory集合。

var cursor = inventory.Watch();
while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
var next = cursor.Current.First();
cursor.Dispose();

以下Go语言示例假设您已连接到MongoDB副本集并访问了数据库，并且包含一个inventory集合。

cs, err := coll.Watch(ctx, mongo.Pipeline{})
assert.NoError(t, err)
defer cs.Close(ctx)
ok := cs.Next(ctx)
next := cs.Current

以下Java示例假设您已经连接到MongoDB副本集并访问了一个数据库，其中包含一个库存集合。

MongoCursor<ChangeStreamDocument<Document>> cursor = inventory.watch().iterator();
ChangeStreamDocument<Document> next = cursor.next();

以下Kotlin示例假设您已连接到MongoDB副本集并可以访问包含库存集合的数据库。要了解更多关于完成这些任务的信息，请参阅Kotlin Coroutine Driver 数据库和集合指南。

val job = launch {
    val changeStream = collection.watch()
    changeStream.collect {
        println("Received a change event: $it")
    }
}

以下示例假设您已经连接到MongoDB副本集并访问了一个数据库，其中包含一个库存集合。

cursor = db.inventory.watch()
document = await cursor.next()

以下Node.js示例假设您已连接到MongoDB副本集并访问了一个数据库，其中包含一个库存集合。

以下示例使用流来处理更改事件。

const collection = db.collection('inventory');
const changeStream = collection.watch();
changeStream.on('change', next => {
  // process next document
});

或者，您也可以使用迭代器来处理更改事件

const collection = db.collection('inventory');
const changeStream = collection.watch();
const next = await changeStream.next();

ChangeStream扩展自EventEmitter。

以下示例假设您已连接到MongoDB副本集并访问了一个数据库，其中包含一个库存集合。

$changeStream = $db->inventory->watch();
$changeStream->rewind();
$firstChange = $changeStream->current();
$changeStream->next();
$secondChange = $changeStream->current();

以下Python示例假设您已连接到MongoDB副本集并访问了一个数据库，其中包含一个库存集合。

cursor = db.inventory.watch()
next(cursor)

以下示例假设您已连接到MongoDB副本集并访问了一个数据库，其中包含一个库存集合。

cursor = inventory.watch.to_enum
next_change = cursor.next

以下Swift（异步）示例假设你已经连接到MongoDB副本集并访问了一个数据库，该数据库包含一个库存集合。（点击此处查看详细说明）

let inventory = db.collection("inventory")
// Option 1: retrieve next document via next()
let next = inventory.watch().flatMap { cursor in
    cursor.next()
}
// Option 2: register a callback to execute for each document
let result = inventory.watch().flatMap { cursor in
    cursor.forEach { event in
        // process event
        print(event)
    }
}

以下Swift（同步）示例假设你已经连接到MongoDB副本集并访问了一个数据库，该数据库包含一个库存集合。（点击此处查看详细说明）

let inventory = db.collection("inventory")
let changeStream = try inventory.watch()
let next = changeStream.next()

要从游标中检索数据变更事件，请迭代变更流游标。有关变更流事件的更多信息，请参阅变更事件。

以下条件之一发生时，变更流游标将保持打开状态：

游标被显式关闭。
发生无效化事件；例如，集合删除或重命名。
与MongoDB部署的连接关闭或超时。有关更多信息，请参阅游标行为。
如果部署是分片集群，移除分片可能会使打开的变更流游标关闭。关闭的变更流游标可能无法完全恢复。

注意

未关闭游标的生命周期依赖于语言。

[1]	你可以指定一个`startAtOperationTime`来在特定时间点打开游标。如果指定的起始点在过去，它必须在oplog的时间范围内。