监控数据变化

Overview

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

使用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字段，您可以通过调用相应的 setter 方法来设立这些字段：

• add_fields：向文档添加新字段

• match：筛选文档

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

• redact：限制文档内容

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

• merge：将结果输出到集合

以下示例设置mongocxx::pipeline实例的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::find实例中设立的字段：

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

更多信息

要了解有关变更流的更多信息，请参阅Change Streams MongoDB Server手册中的 。