Change Streams

在此页面上

可用性

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

MongoDB5.0 已于 10 月2024 结束生命周期。不再支持此版本的文档。要升级5.0 部署，请参阅 MongoDB6 。0 升级程序。

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

可用性

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

存储引擎。
副本集和分片集群必须使用 WiredTiger 存储引擎。 Change stream 也可以用于采用 MongoDB 的静态加密功能的部署。
副本集协议版本。
副本集和分片集群必须使用副本集协议版本 1 (pv1)。
读关注“多数”启用。
从 MongoDB 4.2 开始，无论是否有 "majority" 读关注支持，change stream 都可用；也就是说，读关注majority 支持可以启用（默认），也可以禁用以使用 change stream。
在 MongoDB 4.0 及更早版本中，变更流仅在系统启用 "majority" 读关注支持（默认）的情况下才可用。

稳定的 API 支持

变更流包含在Stable API V1 中。

连接

连接 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 设置。

打开变更流

要打开变更流：

对于副本集，您可以从任何承载数据的成员发出打开 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();

以下示例假设您已连接到 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();

以下示例假设您已连接到 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 的时间范围内。