监控数据变化

Overview

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

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

样本数据

本指南中的示例使用 Atlas示例数据集的sample_restaurants数据库中的restaurants集合。要从PHP应用程序访问权限此集合，请实例化一个连接到Atlas 集群的MongoDB\Client ，并将以下值分配给$collection变量：

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

提示

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

一些示例使用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();
for ($changeStream->rewind(); true; $changeStream->next()) {
    if ( ! $changeStream->valid()) {
        continue;
    }
    $event = $changeStream->current();
    echo toJSON($event), PHP_EOL;
    if ($event['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);
for ($changeStream->rewind(); true; $changeStream->next()) {
    if ( ! $changeStream->valid()) {
        continue;
    }
    $event = $changeStream->current();
    echo toJSON($event), PHP_EOL;
    if ($event['operationType'] === 'invalidate') {
        break;
    }
}

提示

与构建者合作

您可以使用构建者模式来创建变更流管道。要学习；了解详情，请参阅构建者操作指南。

修改`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`	Instructs `watch()` to start a new change stream after the operation specified in the resume token. This field allows notifications to resume after an invalidate event. Each change stream event document includes a resume token as the `_id` field. Pass the entire `_id` field of the change event document that represents the operation you want to resume after. This option is mutually exclusive with `resumeAfter` and `startAtOperationTime`.
`startAtOperationTime`	Instructs the change stream to only provide changes that occurred at or after the specified timestamp. This option is mutually exclusive with `startAfter` and `resumeAfter`.
`collation`	Sets the collation to use for the change stream cursor.

有关watch()选项的完整列表，请参阅API文档中的MongoDB \Collection::watch() 。

包含前像和后像

重要

仅当您的部署使用 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);
for ($changeStream->rewind(); true; $changeStream->next()) {
    if ( ! $changeStream->valid()) {
        continue;
    }
    $event = $changeStream->current();
    echo toJSON($event), PHP_EOL;
    if ($event['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手册中的具有文档前图像和后图像的。