/ /

监控数据变化

Overview

在本指南中，您可以学习；了解如何使用变更流来监控数据的实时变更。变更流是MongoDB Server的一项功能，允许应用程序订阅集合、数据库或部署上的数据更改。

提示

Atlas Stream Processing

As an alternative to change streams, you can use Atlas Stream Processing to process and transform streams of data. Unlike change streams, which register only database events, Atlas Stream Processing manages multiple data event types and provides extended data processing capabilities. To learn more about this feature, see Atlas Stream Processing in the MongoDB Atlas documentation.

样本数据

本指南中的示例使用Atlas示例数据集中的 sample_restaurants.restaurants集合。要学习；了解如何创建免费的MongoDB Atlas 群集并加载示例数据集，请参阅.NET/ C#驱动程序入门。

本页的示例使用以下 Restaurant、Address 和 GradeEntry 类作为模型：

public class Restaurant
{
    public ObjectId Id { get; set; }
    public string Name { get; set; }
    [BsonElement("restaurant_id")]
    public string RestaurantId { get; set; }
    public string Cuisine { get; set; }
    public Address Address { get; set; }
    public string Borough { get; set; }
    public List<GradeEntry> Grades { get; set; }
}

public class Address
{
    public string Building { get; set; }
    [BsonElement("coord")]
    public double[] Coordinates { get; set; }
    public string Street { get; set; }
    [BsonElement("zipcode")]
    public string ZipCode { get; set; }
}

public class GradeEntry
{
    public DateTime Date { get; set; }
    public string Grade { get; set; }
    public float? Score { get; set; }
}

注意

restaurants集合中的文档使用蛇形命名规则。本指南中的示例使用 ConventionPack 将集合中的字段反序列化为 Pascal 语句，并将它们映射到 Restaurant 类中的属性。

如需了解有关自定义序列化的更多信息，请参阅“自定义序列化”。

打开变更流

要打开变更流，请调用Watch()或WatchAsync()方法。调用该方法的实例决定了变更流侦听的事件范围。您可以对以下类调用Watch()或WatchAsync()方法：

MongoClient：监控 MongoDB 部署中的所有更改
Database：监控数据库中所有集合的变更
Collection：监控集合中的更改

以下示例在restaurants集合上打开变更流，并在发生变更时输出变更。选择Synchronous或Asynchronous标签页以查看相应的代码。

var database = client.GetDatabase("sample_restaurants");
var collection = database.GetCollection<Restaurant>("restaurants");
// Opens a change stream and prints the changes as they're received
using (var cursor = collection.Watch())
{
    foreach (var change in cursor.ToEnumerable())
    {
        Console.WriteLine("Received the following type of change: " + change.BackingDocument);
    }
}

var database = client.GetDatabase("sample_restaurants");
var collection = database.GetCollection<Restaurant>("restaurants");
// Opens a change streams and print the changes as they're received
using var cursor = await collection.WatchAsync();
await cursor.ForEachAsync(change =>
{
    Console.WriteLine("Received the following type of change: " + change.BackingDocument);
});

要开始监视更改，请运行应用程序。然后，在单独的应用程序或shell中，修改 restaurants集合。更新"name"值为"Blarney Castle"的文档会产生以下变更流输出：

{ "_id" : { "_data" : "..." }, "operationType" : "update", "clusterTime" : Timestamp(...),
"wallTime" : ISODate("..."), "ns" : { "db" : "sample_restaurants", "coll" : "restaurants" },
"documentKey" : { "_id" : ObjectId("...") }, "updateDescription" : { "updatedFields" : { "cuisine" : "Irish" },
"removedFields" : [], "truncatedArrays" : [] } }

修改变更流输出

您可以将pipeline参数传递给Watch()和WatchAsync()方法，以修改变更流输出。此参数允许您仅监视指定的更改事件。使用EmptyPipelineDefinition类并附加相关聚合阶段方法来创建管道。

您可以在pipeline参数中指定以下聚合阶段：

$addFields
$changeStreamSplitLargeEvent
$match
$project
$replaceRoot
$replaceWith
$redact
$set
$unset

提示

要学习；了解如何使用 PipelineDefinitionBuilder 类构建聚合管道，请参阅 Operations with Builders指南中的聚合管道阶段。

To learn more about modifying your change stream output, see the Modify Change Stream Output section in the MongoDB Server manual.

监控更新事件示例

以下示例使用pipeline参数打开仅记录更新操作的变更流。选择Synchronous或Asynchronous标签页以查看相应的代码。

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
            .Match(change => change.OperationType == ChangeStreamOperationType.Update);
// Opens a change streams and print the changes as they're received
using (var cursor = collection.Watch(pipeline))
{
    foreach (var change in cursor.ToEnumerable())
    {
        Console.WriteLine("Received the following change: " + change);
    }
}

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
            .Match(change => change.OperationType == ChangeStreamOperationType.Update);
// Opens a change stream and prints the changes as they're received
using (var cursor = await collection.WatchAsync(pipeline))
{
    await cursor.ForEachAsync(change =>
    {
        Console.WriteLine("Received the following change: " + change);
    });
}

分割大型更改事件示例

如果应用应用程序生成大小超过 16 MB 的变更事件，服务器将返回 BSONObjectTooLarge 错误。为避免此错误，您可以使用 $changeStreamSplitLargeEvent管道阶段将事件分割为较小的片段。 .NET/ C#驱动程序聚合API包括 ChangeStreamSplitLargeEvent() 方法，您可以使用该方法将 $changeStreamSplitLargeEvent 阶段添加到变更流管道。

此示例指示驱动程序监视超过 16 MB 限制的更改和分割更改事件。该代码会打印每个事件的变更文档，并调用辅助工具来重新组合所有事件片段：

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .ChangeStreamSplitLargeEvent();
using var cursor = collection.Watch(pipeline);
foreach (var completeEvent in GetNextChangeStreamEvent(cursor.ToEnumerable().GetEnumerator()))
{
    Console.WriteLine("Received the following change: " + completeEvent.BackingDocument);
}

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .ChangeStreamSplitLargeEvent();
using var cursor = await collection.WatchAsync(pipeline);
await foreach (var completeEvent in GetNextChangeStreamEventAsync(cursor))
{
    Console.WriteLine("Received the following change: " + completeEvent.BackingDocument);
}

注意

我们建议重新组合变更事件片段，如前面的示例所示，但这一步是可选的。您可以使用相同的逻辑来监视分割和完成变更事件。

前面的示例使用 GetNextChangeStreamEvent()、GetNextChangeStreamEventAsync() 和 MergeFragment() 方法将变更事件片段重新组合成单个变更流文档。以下代码定义了这些方法：

// Fetches the next complete change stream event
private static IEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEvent<TDocument>(
IEnumerator<ChangeStreamDocument<TDocument>> changeStreamEnumerator)
{
    while (changeStreamEnumerator.MoveNext())
    {
        var changeStreamEvent = changeStreamEnumerator.Current;
        if (changeStreamEvent.SplitEvent != null)
        {
            var fragment = changeStreamEvent;
            while (fragment.SplitEvent.Fragment < fragment.SplitEvent.Of)
            {
                changeStreamEnumerator.MoveNext();
                fragment = changeStreamEnumerator.Current;
                MergeFragment(changeStreamEvent, fragment);
            }
        }
        yield return changeStreamEvent;
    }
}
// Merges a fragment into the base event
private static void MergeFragment<TDocument>(
    ChangeStreamDocument<TDocument> changeStreamEvent,
    ChangeStreamDocument<TDocument> fragment)
{
    foreach (var element in fragment.BackingDocument)
    {
        if (element.Name != "_id" && element.Name != "splitEvent")
        {
            changeStreamEvent.BackingDocument[element.Name] = element.Value; 
        }
    }
}

// Fetches the next complete change stream event
private static async IAsyncEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEventAsync<TDocument>(
    IAsyncCursor<ChangeStreamDocument<TDocument>> changeStreamCursor)
{
    var changeStreamEnumerator = GetNextChangeStreamEventFragmentAsync(changeStreamCursor).GetAsyncEnumerator();
    while (await changeStreamEnumerator.MoveNextAsync())
    {
        var changeStreamEvent = changeStreamEnumerator.Current;
        if (changeStreamEvent.SplitEvent != null)
        {
            var fragment = changeStreamEvent;
            while (fragment.SplitEvent.Fragment < fragment.SplitEvent.Of)
            {
                await changeStreamEnumerator.MoveNextAsync();
                fragment = changeStreamEnumerator.Current;
                MergeFragment(changeStreamEvent, fragment);
            }
        }
        yield return changeStreamEvent;
    }
}
private static async IAsyncEnumerable<ChangeStreamDocument<TDocument>> GetNextChangeStreamEventFragmentAsync<TDocument>(
    IAsyncCursor<ChangeStreamDocument<TDocument>> changeStreamCursor)
{
    while (await changeStreamCursor.MoveNextAsync())
    {
        foreach (var changeStreamEvent in changeStreamCursor.Current)
        {
            yield return changeStreamEvent;
        }
    }
}
// Merges a fragment into the base event
private static void MergeFragment<TDocument>(
    ChangeStreamDocument<TDocument> changeStreamEvent,
    ChangeStreamDocument<TDocument> fragment)
{
    foreach (var element in fragment.BackingDocument)
    {
        if (element.Name != "_id" && element.Name != "splitEvent")
        {
            changeStreamEvent.BackingDocument[element.Name] = element.Value;
        }
    }
}

提示

To learn more about splitting large change events, see $changeStreamSplitLargeEvent in the MongoDB Server manual.

修改 `Watch()` 行为

Watch() 和 WatchAsync() 方法接受可选参数，这些参数表示可用于配置操作的选项。如果不指定任何选项，驱动程序不会自定义操作。

下表描述了可以设立用于自定义Watch()和WatchAsync()行为的选项：

选项	说明
`FullDocument`	Specifies whether to show the full document after the change, rather than showing only the changes made to the document. To learn more about this option, see Include Pre-Images and Post-Images.
`FullDocumentBeforeChange`	Specifies whether to show the full document as it was before the change, rather than showing only the changes made to the document. To learn more about this option, see Include Pre-Images and Post-Images.
`ResumeAfter`	指示 `Watch()` 或 `WatchAsync()` 在恢复令牌中指定的操作之后恢复返回变更。每份变更流事件文档都包含一个恢复令牌作为 `_id` 字段。传递代表要恢复的操作的变更事件文档的整个 `_id` 字段。 `ResumeAfter` 与 `StartAfter` 和 `StartAtOperationTime` 互斥。
`StartAfter`	指示 `Watch()` 或 `WatchAsync()` 在恢复令牌中指定的操作后启动新的变更流。允许在无效事件后恢复通知。每份变更流事件文档都包含一个恢复令牌作为 `_id` 字段。传递代表要恢复的操作的变更事件文档的整个 `_id` 字段。 `StartAfter` 与 `ResumeAfter` 和 `StartAtOperationTime` 互斥。
`StartAtOperationTime`	指导 `Watch()` 或 `WatchAsync()` 仅返回在指定时间戳后发生的事件。 `StartAtOperationTime` 与 `ResumeAfter` 和 `StartAfter` 互斥。
`MaxAwaitTime`	指定服务器在返回空批处理之前等待新数据更改报告给变更流游标的最长时间（以毫秒为单位）。默认为 1000 毫秒。
`ShowExpandedEvents`	从 MongoDB Server v 6.0开始，变更流支持数据定义语言 (DDL) 事件的变更通知，例如`createIndexes`和`dropIndexes`事件。要在变更流中包含扩展事件，请创建变更流游标并将此参数设置为`True` 。
`batchSize`	指定变更流在每个批处理中可返回的最大文档数量，这适用于 `Watch()` 或 `WatchAsync()`。如果未设置 `batchSize` 选项，则 watch 函数的初始批处理大小为 `101` 个文档，后续批处理的最大大小为 16 梅比字节 (MiB)。此选项可以强制比 16 MiB 更小的限制，但不能强制更大的限制。如果将 `batchSize` 设置为导致批处理大于 16 MiB 的限制，则此选项无效，`Watch()` 或 `WatchAsync()` 使用默认批处理大小。
`Collation`	Specifies the collation to use for the change stream cursor. See the Collation section of this page for more information.
`Comment`	为操作附加注释。

排序规则

要为操作配置排序规则，请创建 Collation 类的实例。

下表描述了 Collation 构造函数接受的参数。它还列出了相应的类属性，您可以使用这些属性读取每个设置的值。

Parameter	说明	类属性
`locale`	Specifies the International Components for Unicode (ICU) locale. For a list of supported locales, see Collation Locales and Default Parameters in the MongoDB Server Manual. If you want to use simple binary comparison, use the `Collation.Simple` static property to return a `Collation` object with the `locale` set to `"simple"`. Data Type: `string`	`Locale`
`caseLevel`	（可选）指定是否包含大小写比较。当此参数为 `true` 时，驱动程序的行为取决于 `strength` 参数的值： - 如果 `strength` 为 `CollationStrength.Primary`，则驱动程序会比较基本字符和大小写。 - 如果 `strength` 为 `CollationStrength.Secondary`，则驱动程序会比较基本字符、变音符、其他次要差异和大小写。 - 如果 `strength` 为任何其他值，则会忽略此参数。当此参数为 `false` 时，驱动程序不会在强度级别 `Primary` 或 `Secondary` 包含大小写比较。数据类型：`boolean` 默认值：`false`	`CaseLevel`
`caseFirst`	（可选）指定三级比较期间大小写差异的排序顺序。数据类型：CollationCaseFirst 默认值：`CollationCaseFirst.Off`	`CaseFirst`
`strength`	(可选) 指定要执行的比较级别，如 ICU 文档中所定义。数据类型：CollationStrength 默认：`CollationStrength.Tertiary`	`Strength`
`numericOrdering`	(Optional) Specifies whether the driver compares numeric strings as numbers. If this argument is `true`, the driver compares numeric strings as numbers. For example, when comparing the strings "10" and "2", the driver treats the values as 10 and 2, and finds 10 to be greater. If this argument is `false` or excluded, the driver compares numeric strings as strings. For example, when comparing the strings "10" and "2", the driver compares one character at a time. Because "1" is less than "2", the driver finds "10" to be less than "2". For more information, see Collation Restrictions in the MongoDB Server manual. Data Type: `boolean` Default: `false`	`NumericOrdering`
`alternate`	(可选) 指定驱动程序是否将空格和标点视为比较的基本字符。数据类型：CollationAlternate 默认值：`CollationAlternate.NonIgnorable` （空格和标点被视为基本字符）	`Alternate`
`maxVariable`	（可选）指定当 `alternate` 参数为 `CollationAlternate.Shifted` 时，驱动程序认为可忽略的字符。数据类型：CollationMaxVariable 默认：`CollationMaxVariable.Punctuation`（驱动程序忽略标点符号和空格）	`MaxVariable`
`normalization`	（可选）指定驱动程序是否根据需要规范化文本。大多数文本不需要规范化。有关规范化的更多信息，请参阅 ICU 文档。数据类型：`boolean` 默认：`false`	`Normalization`
`backwards`	（可选）指定包含变音符号的字符串是否从字符串的后部到前部排序。数据类型：`boolean` 默认：`false`	`Backwards`

For more information about collation, see the Collation page in the MongoDB Server manual.

包含前像和后像

重要

仅当您的部署使用 MongoDB v 6.0或更高版本时，才能对集合启用前图像和后图像。

默认，当您对集合执行操作时，相应的变更事件仅包括该操作修改的字段的增量。要查看更改之前或之后的完整文档，请创建一个ChangeStreamOptions对象并指定FullDocumentBeforeChange或FullDocument选项。然后，将ChangeStreamOptions对象传递给Watch()或WatchAsync()方法。

前像是文档在更改之前的完整版本。要将前像包含在变更流事件，请将FullDocumentBeforeChange选项设立为以下值之一：

ChangeStreamFullDocumentBeforeChangeOption.WhenAvailable：仅当预像可用时，变更事件才包含变更事件的已修改文档的前像。
ChangeStreamFullDocumentBeforeChangeOption.Required：变更事件包括变更事件的已修改文档的前像。如果前像不可用，则驱动程序会引发错误。

后像是文档更改后的完整版本。要将后图像包含在变更流事件，请将FullDocument选项设立为以下值之一：

ChangeStreamFullDocumentOption.UpdateLookup：更改事件包括更改后某个时间点的整个已更改文档的副本。
ChangeStreamFullDocumentOption.WhenAvailable：仅当后图像可用时，更改事件才包含更改事件的已修改文档的后图像。
ChangeStreamFullDocumentOption.Required：变更事件包括变更事件的已修改文档的后像。如果后图像不可用，驱动程序会引发错误。

以下示例在集合上打开变更流，并通过指定FullDocument选项包含更新文档的后映像。选择Synchronous或Asynchronous标签页以查看相应的代码。

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .Match(change => change.OperationType == ChangeStreamOperationType.Update);
var options = new ChangeStreamOptions
{
    FullDocument = ChangeStreamFullDocumentOption.UpdateLookup,
};
using (var cursor = collection.Watch(pipeline, options))
{
    foreach (var change in cursor.ToEnumerable())
    {
        Console.WriteLine(change.FullDocument.ToBsonDocument());
    }
}

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<Restaurant>>()
    .Match(change => change.OperationType == ChangeStreamOperationType.Update);
var options = new ChangeStreamOptions
{
    FullDocument = ChangeStreamFullDocumentOption.UpdateLookup,
};
using var cursor = await collection.WatchAsync(pipeline, options);
await cursor.ForEachAsync(change =>
{
    Console.WriteLine(change.FullDocument.ToBsonDocument());
});

运行前面的代码示例并更新"name"值为"Blarney Castle"的文档会产生以下变更流输出：

{ "_id" : ObjectId("..."), "name" : "Blarney Castle", "restaurant_id" : "40366356",
"cuisine" : "Traditional Irish", "address" : { "building" : "202-24", "coord" : [-73.925044200000002, 40.5595462],
"street" : "Rockaway Point Boulevard", "zipcode" : "11697" }, "borough" : "Queens", "grades" : [...] }

要了解有关前图像和后图像的更多信息，请参阅Change Streams MongoDB Server手册中的具有文档前图像和后图像的。