更改流_MonogDB 中文网

参考 > 更改流

在本页面

可用性
监视集合/数据库/部署
打开变更流
修改变更流输出
查找完整文档以进行更新操作
恢复变更流
用例
访问控制
活动通知
校对

3.6版的新功能。

变更流允许应用程序访问实时数据变更，而不会带来复杂性和拖延oplog的风险。应用程序可以使用更改流来订阅单个集合，数据库或整个部署上的所有数据更改，并立即对其做出反应。由于变更流使用聚合框架，因此应用程序还可以过滤特定变更或随意转换通知。

可用性¶

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

存储引擎。

副本集和分片群集必须使用WiredTiger存储引擎。变更流还可以用于采用MongoDB静态加密功能的部署。
副本集协议版本。

副本集和分片群集必须使用副本集协议版本1（pv1）。
阅读关注“多数”启用。

从MongoDB 4.2开始，无论是否支持读关注，更改流都可用"majority"。也就是说，majority可以启用阅读关注支持（默认）或禁用以使用更改流。

在MongoDB 4.0和更早版本中，更改流仅在"majority"启用了阅读关注支持时才可用（默认）。

监视集合/数据库/部署¶

您可以针对以下情况打开变更流：

目标	描述
集合	你可以打开一个变换流光标单个集合（除`system`收藏，或任何集合在 `admin`，`local`和`config`数据库）。此页面上的示例使用MongoDB驱动程序打开并使用单个集合的变更流游标。另请参见`mongo`shell方法 `db.collection.watch()`。
数据库	在MongoDB中4.0开始，你可以打开一个变换流光标单个数据库（不包括`admin`，`local`和 `config`数据库）来监视更改其所有非系统集合。有关MongoDB驱动程序方法，请参阅驱动程序文档。另请参见`mongo`shell方法 `db.watch()`。
部署	在MongoDB中4.0开始，你可以打开一个变换流光标部署（无论是副本集中或分片集群）来监视除了所有数据库中更改所有非系统集合`admin`，`local`和`config`。有关MongoDB驱动程序方法，请参阅驱动程序文档。另请参见`mongo`shell方法 `Mongo.watch()`。

更改流示例

此页面上的示例使用MongoDB驱动程序来说明如何打开集合的更改流游标以及如何使用更改流游标。

打开变更流¶

要打开变更流：

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

以下示例打开一个集合的变更流，并在光标上进行迭代以检索变更流文档。 [1]

下面的Python示例假定您已连接到MongoDB副本集并访问了包含inventory集合的数据库。

复制

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

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

复制

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

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

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

复制

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

另外，您还可以使用迭代器来处理更改事件：

复制

const changeStreamIterator = collection.watch();
const next = await changeStreamIterator.next();

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

复制

$changeStream = $db->inventory->watch();
$changeStream->rewind();

$firstChange = $changeStream->current();

$changeStream->next();

$secondChange = $changeStream->current();

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

复制

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

下面的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();

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

复制

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

下面的Go示例假定您已连接到MongoDB副本集并访问了包含inventory集合的数据库。

复制

cs, err := coll.Watch(ctx, mongo.Pipeline{})
require.NoError(t, err)
defer cs.Close(ctx)

ok := cs.Next(ctx)
next := cs.Current

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

与MongoDB部署的连接保持打开状态时，游标保持打开状态，直到发生以下情况之一：

游标已明确关闭。
发生无效事件。
如果部署是分片群集，则分片删除可能会导致打开的更改流游标关闭，并且关闭的更改流游标可能无法完全恢复。

注意

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

[1]	从MongoDB 4.0开始，您可以指定a `startAtOperationTime` 在特定时间点打开游标。如果指定的起点是过去的时间，则必须在操作日志的时间范围内。

修改变更流输出¶

在配置变更流时，可以通过提供以下一个或多个以下管道阶段的数组来控制变更流的输出：

$addFields
$match
$project
$replaceRoot
$replaceWith （从MongoDB 4.2开始可用）
$redact
$set （从MongoDB 4.2开始可用）
$unset （从MongoDB 4.2开始可用）

复制

pipeline = [
    {'$match': {'fullDocument.username': 'alice'}},
    {'$addFields': {'newField': 'this is an added field!'}}
]
cursor = db.inventory.watch(pipeline=pipeline)
document = next(cursor)

在配置变更流时，可以通过提供以下一个或多个以下管道阶段的数组来控制变更流的输出：

$addFields
$match
$project
$replaceRoot
$replaceWith （从MongoDB 4.2开始可用）
$redact
$set （从MongoDB 4.2开始可用）
$unset （从MongoDB 4.2开始可用）

复制

MongoClient mongoClient = new MongoClient( new MongoClientURI("mongodb://host1:port1,host2:port2..."));

// Select the MongoDB database and collection to open the change stream against

MongoDatabase db = mongoClient.getDatabase("myTargetDatabase");

MongoCollection<Document> collection = db.getCollection("myTargetCollection");

// Create $match pipeline stage.
List<Bson> pipeline = singletonList(Aggregates.match(Filters.or(
    Document.parse("{'fullDocument.username': 'alice'}"),
    Filters.in("operationType", asList("delete")))));

// Create the change stream cursor, passing the pipeline to the
// collection.watch() method

MongoCursor<Document> cursor = collection.watch(pipeline).iterator();

所述pipeline列表包括单个$match，其过滤的任何操作，其中阶段username是alice，或操作，其中operationType是delete。

将传递pipeline给watch()方法后，将更改流传递给通知，使其通过指定的pipeline。

在配置变更流时，可以通过提供以下一个或多个以下管道阶段的数组来控制变更流的输出：

$addFields
$match
$project
$replaceRoot
$replaceWith （从MongoDB 4.2开始可用）
$redact
$set （从MongoDB 4.2开始可用）
$unset （从MongoDB 4.2开始可用）

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

复制

const pipeline = [
  { $match: { 'fullDocument.username': 'alice' } },
  { $addFields: { newField: 'this is an added field!' } }
];

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

另外，您还可以使用迭代器来处理更改事件：

复制

const changeStreamIterator = collection.watch(pipeline);
const next = await changeStreamIterator.next();

在配置变更流时，可以通过提供以下一个或多个以下管道阶段的数组来控制变更流的输出：

$addFields
$match
$project
$replaceRoot
$replaceWith （从MongoDB 4.2开始可用）
$redact
$set （从MongoDB 4.2开始可用）
$unset （从MongoDB 4.2开始可用）

复制

$pipeline = [
    ['$match' => ['fullDocument.username' => 'alice']],
    ['$addFields' => ['newField' => 'this is an added field!']],
];
$changeStream = $db->inventory->watch($pipeline);
$changeStream->rewind();

$firstChange = $changeStream->current();

$changeStream->next();

$secondChange = $changeStream->current();

在配置变更流时，可以通过提供以下一个或多个以下管道阶段的数组来控制变更流的输出：

$addFields
$match
$project
$replaceRoot
$replaceWith （从MongoDB 4.2开始可用）
$redact
$set （从MongoDB 4.2开始可用）
$unset （从MongoDB 4.2开始可用）

复制

pipeline = [
    {'$match': {'fullDocument.username': 'alice'}},
    {'$addFields': {'newField': 'this is an added field!'}}
]
cursor = db.inventory.watch(pipeline=pipeline)
document = await cursor.next()

在配置变更流时，可以通过提供以下一个或多个以下管道阶段的数组来控制变更流的输出：

$addFields
$match
$project
$replaceRoot
$replaceWith （从MongoDB 4.2开始可用）
$redact
$set （从MongoDB 4.2开始可用）
$unset （从MongoDB 4.2开始可用）

复制

pipeline = BCON_NEW ("pipeline",
                     "[",
                     "{",
                     "$match",
                     "{",
                     "fullDocument.username",
                     BCON_UTF8 ("alice"),
                     "}",
                     "}",
                     "{",
                     "$addFields",
                     "{",
                     "newField",
                     BCON_UTF8 ("this is an added field!"),
                     "}",
                     "}",
                     "]");

stream = mongoc_collection_watch (collection, pipeline, &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);

在配置变更流时，可以通过提供以下一个或多个以下管道阶段的数组来控制变更流的输出：

$addFields
$match
$project
$replaceRoot
$replaceWith （从MongoDB 4.2开始可用）
$redact
$set （从MongoDB 4.2开始可用）
$unset （从MongoDB 4.2开始可用）

复制

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match(change =>
        change.FullDocument["username"] == "alice" ||
        change.OperationType == ChangeStreamOperationType.Delete)
    .AppendStage<ChangeStreamDocument<BsonDocument>, ChangeStreamDocument<BsonDocument>, BsonDocument>(
        "{ $addFields : { newField : 'this is an added field!' } }");

var collection = database.GetCollection<BsonDocument>("inventory");
using (var cursor = collection.Watch(pipeline))
{
    while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
    var next = cursor.Current.First();
}

在配置变更流时，可以通过提供以下一个或多个以下管道阶段的数组来控制变更流的输出：

$addFields
$match
$project
$replaceRoot
$replaceWith （从MongoDB 4.2开始可用）
$redact
$set （从MongoDB 4.2开始可用）
$unset （从MongoDB 4.2开始可用）

你可以控制

MongoDB 中文手册

更改流

可用性¶

监视集合/数据库/部署¶

打开变更流¶

修改变更流输出¶