监控数据变化

概述

在本指南中，您可以学习如何使用 更改流 来监视数据库的实时更改。更改流是 MongoDB 服务器的一个功能，允许您的应用程序订阅集合、数据库或部署上的数据更改。

当使用 C++ 驱动程序时，您可以实例化一个mongocxx::change_stream 来监视数据更改。

auto db = client["sample_restaurants"];
auto collection = db["restaurants"];

打开变更流

要打开变更流，请调用 watch() 方法。你调用 watch() 方法时所在的实例决定了变更流监听的事件范围。你可以在以下类上调用 watch() 方法

• mongocxx::client：监视MongoDB部署中的所有更改

• mongocxx::database：监视数据库中所有集合的更改

• mongocxx::collection：监视集合的更改

以下示例在 restaurants 集合上打开变更流，并按发生顺序输出更改

auto stream = collection.watch();
while (true) {
    for (const auto& event : stream) {
        std::cout << bsoncxx::to_json(event) << std::endl;
    }
}

要开始监听更改，运行前面的代码。然后，在单独的应用程序或shell中修改 restaurants 集合。以下示例更新一个具有 name 字段值为 Blarney Castle 的文档

auto result = collection.update_one(make_document(kvp("name", "Blarney Castle")), 
                                    make_document(kvp("$set", 
                                    make_document(kvp("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" : [  ] } }

修改变更流输出

你可以通过将 mongocxx::pipeline 实例作为参数传递给 watch() 方法来修改变更流输出。以下列表包括你可以通过调用它们的相应设置方法设置的某些 mongocxx::pipeline 字段

• add_fields：向文档添加新字段

• match：过滤文档

• project：投影文档的字段子集

• redact：限制文档内容

• group：按指定的表达式对文档进行分组

• merge：将结果输出到集合

以下示例设置 match 字段，然后将管道传递给 watch() 方法。这将指示 watch() 方法仅输出更新操作

mongocxx::pipeline pipeline;
pipeline.match(make_document(kvp("operationType", "update")));
auto stream = collection.watch(pipeline);
while (true) {
    for (const auto& event : stream) {
        std::cout << bsoncxx::to_json(event) << std::endl;
    }
}

修改 watch() 行为

您可以通过传递 mongocxx::options::change_stream 类的实例作为参数来修改 watch() 方法的行为。以下表格描述了您可以在 mongocxx::options::change_stream 实例中设置的字段

mongocxx::options::change_stream opts;
opts.full_document("updateLookup");
auto stream = collection.watch(opts);
while (true) {
    for (const auto& event : stream) {
        std::cout << bsoncxx::to_json(event) << std::endl;
    }
}

{ "_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" : [  ] } }

更多信息

要了解更多关于更改流的信息，请参阅 MongoDB 服务器手册中的 更改流。