Change Streams

在此页面上

可用性

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

变更流允许应用程序访问实时数据更改，从而避免事先手动追踪 oplog 的复杂性和风险。应用程序可使用变更流来订阅针对单个集合、数据库或整个部署的所有数据变更，并立即对它们做出响应。由于变更流采用聚合框架，因此，应用程序还可对特定变更进行过滤，或是随意转换通知。

从 MongoDB 5.1 开始，我们对变更流进行了优化，提高了资源利用率，并加快了某些聚合管道阶段的处理速度。

可用性

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

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

稳定的 API 支持

变更流包含在Stable API V1 中。但是， Stable APIV 中不包含 showExpandedEvents 选项。1

连接

连接 change stream 可以使用带有+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 查询的使用。

打开变更流

要打开变更流：

对于副本集，您可以从任何承载数据的成员发出打开 change stream 操作。
对于分片集群，必须从 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 副本集，并且已访问一个包含 inventory 集合的数据库。

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

下面的Kotlin示例假设您已连接到MongoDB副本集，并且可以访问权限包含inventory集合的数据库。要学习；了解有关完成这些任务的更多信息，请参阅Kotlin协程驱动程序数据库和集合指南。

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

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

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

以下 Node.js 示例假设您已连接到 MongoDB 副本集，并且已访问包含 inventory 集合的数据库。

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

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 副本集，并且已访问一个包含 inventory 集合的数据库。

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

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

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

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

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

下面的 Swift (异步) 示例假设您已连接到 MongoDB 副本集，并已访问包含 inventory 集合的数据库。

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 副本集，并已访问包含 inventory 集合的数据库。

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

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

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

游标已明确关闭。
发生失效事件；例如删除或重命名集合。
与 MongoDB 部署之间的连接关闭或超时。有关更多信息，请参阅游标行为。
如果部署是分片集群，则分片删除可能会导致打开的变更流游标关闭。关闭的变更流游标可能无法完全恢复。

注意

未关闭游标的生命周期取决于语言。

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