/ /

使用变更流监控数据

Overview

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

使用MongoDB PHP库时，您可以调用watch()方法返回MongoDB\ChangeStream的实例。然后，您可以遍历MongoDB\ChangeStream实例以监控数据更改，例如更新、插入和删除。

提示

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.

样本数据

The examples in this guide use the restaurants collection in the sample_restaurants database from the Atlas sample datasets. To access this collection from your PHP application, instantiate a MongoDB\Client that connects to an Atlas cluster and assign the following value to your $collection variable:

$collection = $client->sample_restaurants->restaurants;

提示

要了解如何创建免费的 MongoDB 部署并加载示例数据集，请参阅 MongoDB入门指南。

一些示例使用toJSON()函数将变更事件（属于BSON文档）表示为扩展JSON。要使用此函数，请将以下代码粘贴到应用程序文件：

function toJSON(object $document): string
{
    return MongoDB\BSON\Document::fromPHP($document)->toRelaxedExtendedJSON();
}

打开变更流

要打开变更流，请调用watch()方法。您调用watch()方法的实例决定了变更流监控的事件范围。您可以对以下类的实例调用watch()方法：

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

以下示例在restaurants集合上打开变更流，并在发生变更时输出变更：

$changeStream = $collection->watch();
$changeStream->rewind();
while (true) {
    $changeStream->next();
    if ($changeStream->valid()) {
        continue;
    }
    $event = $changeStream->current();
    echo toJSON($event), PHP_EOL;
    if ($changeStream->current()['operationType'] === 'invalidate') {
        break;
    }
}

要开始监视更改，请运行前面的代码。然后，在另一个shell中修改 restaurants集合。以下示例更新了name字段值为'Blarney Castle'的文档：

$result = $collection->updateOne(
    ['name' => 'Blarney Castle'],
    ['$set' => ['cuisine' => 'Irish']],
);

更新集合时，变更流应用程序会在发生变更时打印变更。打印的变更事件类似于以下输出：

{ "_id" : { "_data" : "..." }, "operationType" : "update", "clusterTime" :
{ "$timestamp" : { ... } }, "wallTime" : { "$date" : "..." }, "ns" : { "db" :
"sample_restaurants", "coll" : "restaurants" }, "documentKey" : { "_id" :
{ "$oid" : "..." } }, "updateDescription" : { "updatedFields" : { "cuisine" : "Irish" },
"removedFields" : [  ], "truncatedArrays" : [  ] } }

修改变更流输出

要修改变更流输出，可以将大量中的管道阶段作为参数传递给watch()方法。您可以在大量中包含以下阶段：

$addFields 或$set ：向文档添加新字段
$match：筛选文档
$project：投影文档字段的子集
$replaceWith 或$replaceRoot ：将输入文档替换为指定文档
$redact：限制文档内容
$unset：从文档中删除字段

以下示例将包含$match阶段的管道传递给watch()方法。这指示watch()方法仅在发生更新操作时输出事件：

$pipeline = [['$match' => ['operationType' => 'update']]];
$changeStream = $collection->watch($pipeline);
$changeStream->rewind();
while (true) {
    $changeStream->next();
    if ($changeStream->valid()) {
        continue;
    }
    $event = $changeStream->current();
    echo toJSON($event), PHP_EOL;
    if ($changeStream->current()['operationType'] === 'invalidate') {
        break;
    }
}

提示

与构建者合作

您可以使用构建者模式来创建变更流管道。要学习；了解更多信息，请参阅《Operations with Builders》指南。

修改 `watch()` 行为

要修改watch()方法的行为，可以将选项大量作为参数传递给watch() 。下表描述了您可以在大量中设立的有用选项：

选项	说明
`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 the Include Pre-Images and Post-Images section of this guide.
`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.
`startAfter`	指示 `watch()` 在恢复令牌中指定的操作后启动新的变更流。此字段允许在无效事件后恢复通知。每份变更流事件文档都包含一个恢复令牌作为 `_id` 字段。传递代表要在其后恢复的操作的变更事件文档的整个 `_id` 字段。此选项与 `resumeAfter` 和 `startAtOperationTime` 互斥。
`startAtOperationTime`	指示变更流仅提供在指定时间戳发生的变更。此选项与 `startAfter` 和 `resumeAfter` 互斥。
`collation`	Sets the collation to use for the change stream cursor. To learn more, see the Collation section of this page.

For a full list of watch() options, see MongoDB\Collection::watch() in the API documentation.

包含前像和后像

重要

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

默认，当您对集合执行操作时，相应的变更事件仅包括该操作修改的字段的增量。要查看更改之前或之后的完整文档，请将大量参数中的fullDocumentBeforeChange或fullDocument选项指定为watch() 。

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

MongoDB\Operation\Watch::FULL_DOCUMENT_BEFORE_CHANGE_WHEN_AVAILABLE：变更事件包括变更事件的已修改文档的前像。如果前像不可用，则此变更事件字段的值为null 。
MongoDB\Operation\Watch::FULL_DOCUMENT_BEFORE_CHANGE_REQUIRED：变更事件包括变更事件的已修改文档的前像。如果前像不可用，服务器将引发错误。

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

MongoDB\Operation\Watch::FULL_DOCUMENT_UPDATE_LOOKUP：更改事件包括更改后某个时间点的整个已更改文档的副本。
MongoDB\Operation\Watch::FULL_DOCUMENT_WHEN_AVAILABLE：变更事件包括变更事件的已修改文档的后像。如果后像不可用，则此变更事件字段的值为null 。
MongoDB\Operation\Watch::FULL_DOCUMENT_REQUIRED：变更事件包括变更事件的已修改文档的后像。如果后图像不可用，服务器将引发错误。

以下示例对集合调用watch()方法，并通过设置fullDocument选项包含更新文档的后像：

$options = ['fullDocument' => MongoDB\Operation\Watch::FULL_DOCUMENT_UPDATE_LOOKUP];
$changeStream = $collection->watch([], $options);
$changeStream->rewind();
while (true) {
    $changeStream->next();
    if ($changeStream->valid()) {
        continue;
    }
    $event = $changeStream->current();
    echo toJSON($event), PHP_EOL;
    if ($changeStream->current()['operationType'] === 'invalidate') {
        break;
    }
}

当变更流应用程序在单独的shell中运行时，使用前面的更新示例更新restaurants集合中的文档会打印类似于以下输出的变更事件：

{ "_id" : { "_data" : "..." }, "operationType" : "update", "clusterTime" :
{ "$timestamp" : { ... } }, "wallTime" : { "$date" : "..." },
"fullDocument" : { "_id" : { "$oid" : "..." }, "address" : { "building" :
"202-24", "coord" : [ -73.925044200000002093, 40.559546199999999772 ], "street"
: "Rockaway Point Boulevard", "zipcode" : "11697" }, "borough" : "Queens",
"cuisine" : "Irish", "grades" : [ ...], "name" : "Blarney Castle", "restaurant_id" :
"40366356" }, "ns" : { "db" : "sample_restaurants", "coll" : "restaurants" },
"documentKey" : { "_id" : { "$oid" : "..." } }, "updateDescription" :
{ "updatedFields" : { "cuisine" : "Irish" }, "removedFields" : [  ],
"truncatedArrays" : [  ] } }

提示

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

排序规则

要为操作指定排序规则，请传递 $options大量参数，该参数将 collation 选项设置为操作方法。将 collation 选项分配给配置排序规则规则的大量。

下表描述了可以设立以配置排序规则的字段：

字段	说明
`locale`	(Required) 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. Data Type: `string`
`caseLevel`	（可选）指定是否包含大小写比较。当设置为`true` 时，比较行为取决于`strength` 字段的值： - 如果`strength` 为，则`1` PHP 库比较基本字符和大小写。 - 如果`strength` 为`2` ，则 PHP 库比较基本字符、变音符、其他从节点差异和大小写。 - 如果`strength` 为任何其他值，则忽略此字段。当设置为 `false` 时，PHP 库不包含强度级别 `1` 或 `2` 的大小写比较。数据类型：`bool` 默认值：`false`
`caseFirst`	（可选）指定三级比较期间大小写差异的排序顺序。数据类型：`string` 默认：`"off"`
`strength`	(可选) 指定要执行的比较级别，如 ICU 文档中所定义。数据类型：`int` 默认：`3`
`numericOrdering`	(Optional) Specifies whether the driver compares numeric strings as numbers. If set to `true`, the PHP library compares numeric strings as numbers. For example, when comparing the strings "10" and "2", the library uses the strings' numeric values and treats "10" as greater than "2". If set to `false`, the PHP library compares numeric strings as strings. For example, when comparing the strings "10" and "2", the library compares one character at a time and treats "10" as less than "2". For more information, see Collation Restrictions in the MongoDB Server manual. Data Type: `bool` Default: `false`
`alternate`	（可选）指定库是否将空格和标点视为基本字符以进行比较。数据类型：`string` 默认：`"non-ignorable"`
`maxVariable`	（可选）指定当 `alternate` 字段设置为 `"shifted"` 时，库认为可忽略的字符。数据类型：`string` 默认值：`"punct"`
`backwards`	（可选）指定包含变音符号的字符串是否从字符串的后部到前部排序。数据类型：`bool` 默认：`false`

要学习；了解有关排序规则和每个字段可能值的更多信息，请参阅MongoDB Server手册中的排序规则条目。

使用变更流监控数据

Overview

样本数据

打开变更流

修改变更流输出

修改 watch() 行为

包含前像和后像

排序规则

更多信息

API 文档

修改 `watch()` 行为