监控数据变化

Overview

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

使用Scala驾驶员时，可以调用 watch() 方法返回 ChangeStreamObservable 的实例。然后，您可以订阅ChangeStreamObservable实例以查看新的数据更改，例如更新、插入和删除。

val database: MongoDatabase = client.getDatabase("sample_restaurants")
val collection: MongoCollection[Document] = database.getCollection("restaurants")

case class LatchedObserver() extends Observer[ChangeStreamDocument[Document]] {
  val latch = new CountDownLatch(1)
  override def onSubscribe(subscription: Subscription): Unit = subscription.request(Long.MaxValue) // Request data
  override def onNext(changeDocument: ChangeStreamDocument[Document]): Unit = println(changeDocument)
  override def onError(throwable: Throwable): Unit = {
      println(s"Error: '$throwable")
      latch.countDown()
  }
  override def onComplete(): Unit = latch.countDown()
  def await(): Unit = latch.await()
}

打开变更流

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

• MongoClient：监控部署中所有数据库中所有集合的更改，不包括系统集合或 admin、local 和 数据库中的集合config

• MongoDatabase：监控一个数据库中所有集合的变更

• MongoCollection：监控对一个集合的更改

以下示例调用 watch() 方法在 restaurants集合上打开变更流。此代码会创建一个 LatchedObserver实例来接收和输出发生的变更：

val observer = LatchedObserver()
collection.watch().subscribe(observer)
observer.await()

要开始监视更改，运行前面的代码。然后，在另一个Shell中运行以下代码，更新name字段值为 "Blarney Castle" 的文档：

val filter = equal("name", "Blarney Castle")
val update = set("cuisine", "American")
collection.updateOne(filter, update)
          .subscribe((res: UpdateResult) => println(res),
                    (e: Throwable) => println(s"There was an error: $e"))

当您运行上述代码来更新集合时，变更流应用程序会在发生变更时打印变更。打印的变更事件类似于以下输出：

ChangeStreamDocument{ operationType=update, resumeToken={"_data": "..."},
namespace=sample_restaurants.restaurants, destinationNamespace=null,
fullDocument=null, fullDocumentBeforeChange=null, documentKey={"_id": {...}},
clusterTime=Timestamp{...}, updateDescription=UpdateDescription{removedFields=[],
updatedFields={"cuisine": "Irish"}, truncatedArrays=[], disambiguatedPaths=null},
txnNumber=null, lsid=null, splitEvent=null, wallTime=BsonDateTime{...}}

修改变更流输出

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

• $addFields 或$set ：向文档添加新字段

• $match：筛选文档

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

• $replaceWith 或$replaceRoot ：将输入文档替换为指定文档

• $redact：限制文档内容

• $unset：从文档中删除字段

Scala驾驶员提供了 Aggregates 类，其中包括用于构建前面的管道阶段的辅助方法。

以下示例创建了一个管道，该管道使用 Aggregates.filter() 方法构建$match 阶段。然后，代码将此管道传递给 watch() 方法，并指示 watch() 仅在发生更新操作时输出事件：

val observer = LatchedObserver()
collection.watch(Seq(Aggregates.filter(Filters.in("operationType", "update"))))
observer.await()

修改 watch() 行为

您可以通过链接 ChangeStreamObservable 类提供的方法来修改 watch() 方法的行为。下表描述了其中一些方法：

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

val observer = LatchedObserver()
collection.watch()
          .fullDocument(FullDocument.UPDATE_LOOKUP)
          .subscribe(observer)
observer.await()

ChangeStreamDocument{ operationType=update, resumeToken={"_data": "..."},
namespace=sample_restaurants.restaurants, destinationNamespace=null,
fullDocument=Iterable((_id,BsonObjectId{...}), (address,{"building": "202-24",
"coord": [-73.9250442, 40.5595462], "street": "Rockaway Point Boulevard",
"zipcode": "11697"}), (borough,BsonString{value='Queens'}),
(cuisine,BsonString{value='Irish'}), (grades,BsonArray{values=[...]}),
(name,BsonString{value='Blarney Castle'}), (restaurant_id,BsonString{...}),
(blank,BsonString{value='Irish'})), fullDocumentBeforeChange=null,
documentKey={"_id": {"$oid": "..."}}, clusterTime=Timestamp{...}, updateDescription=
UpdateDescription{removedFields=[], updatedFields={"cuisine": "Irish"},
truncatedArrays=[], disambiguatedPaths=null}, txnNumber=null, lsid=null,
splitEvent=null, wallTime=BsonDateTime{...}}

更多信息

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