打开变更流

Overview

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

您可以指定一组聚合操作符来筛选和转换应用程序接收的数据。 连接到 MongoDB 部署 v6.0 或更高版本时，您还可以配置事件以包含更改之前和之后的文档数据。

通过以下部分了解如何打开和配置change stream：

• 打开变更流

• 将聚合操作符应用于change stream

• 包含前像和后像

打开变更流

您可以打开变更流来订阅特定类型的数据变更，并在应用程序中生成变更事件。

要打开change stream，请在实例上调用watch() MongoCollectionMongoDatabaseMongoClient方法。

您对其调用watch()方法的对象决定了change stream侦听的事件范围。

如果您对MongoCollection调用watch() ，则change stream会监控一个collection。

如果您对MongoDatabase调用watch() ，则change stream将监控该数据库中的所有collection。

如果您对MongoClient调用watch() ，则change stream会监控已连接 MongoDB 部署中的所有变更。

MongoCollection<Document> collection = database.getCollection("myColl");
ChangeStreamIterable<Document> changeStream = collection.watch();
changeStream.forEach(event ->
        System.out.println("Received a change: " + event));

Received a change: ChangeStreamDocument{
   operationType=insert,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   ...
}

将聚合操作符应用于change stream

您可以将聚合管道作为参数传递给watch()方法，以指定change stream接收哪些事件。

要了解您的 MongoDB Server 版本支持哪些聚合操作符，请参阅修改变更流输出。

MongoCollection<Document> collection = database.getCollection("myColl");
List<Bson> pipeline = Arrays.asList(
        Aggregates.match(Filters.in("operationType", Arrays.asList("insert", "update"))));
ChangeStreamIterable<Document> changeStream = collection.watch(pipeline);
changeStream.forEach(event ->
        System.out.println("Received a change to the collection: " + event));

Received a change: ChangeStreamDocument{
   operationType=update,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   ...
}

分割大型change stream事件

从 MongoDB 7.0 开始，您可以使用$changeStreamSplitLargeEvent聚合阶段将超过 16 MB 的事件分割成更小的片段。

仅在绝对必要时使用$changeStreamSplitLargeEvent 。 例如，如果您的应用程序需要完整的文档前图像或帖子图像，并生成超过 16 MB 的事件，请使用$changeStreamSplitLargeEvent 。

$changeStreamSplitLargeEvent 阶段按顺序返回片段。 您可以使用change stream游标访问这些片段。SplitEvent每个片段都包括一个包含以下字段的对象：

以下示例通过使用$changeStreamSplitLargeEvent聚合阶段分割大型事件来修改change stream：

ChangeStreamIterable<Document> changeStream = collection.watch(
      Arrays.asList(Document.parse("{ $changeStreamSplitLargeEvent: {} }")));

您可以在change stream游标上调用getSplitEvent()方法来访问SplitEvent ，如以下示例所示：

MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = changeStream.cursor();
SplitEvent event = cursor.tryNext().getSplitEvent();

有关$changeStreamSplitLargeEvent聚合阶段的更多信息，请参阅$changeStreamSplitLargeEvent服务器文档。

包含前像和后像

您可以配置变更事件以包含或省略以下数据：

• 前像，表示操作前文档版本的文档（如果存在）

• 帖子-图像，表示操作后文档版本的文档（如果存在）

要接收包含前映像或帖子映像的change stream事件，您必须执行以下操作：

• 为 MongoDB 部署中的collection启用前像和帖子。

  提示

  要了解如何在部署上启用前像和后像，请参阅服务器手册中的带文档前像和后像的变更流。

  要了解如何指示驱动程序创建启用了前图像和后图像的集合，请参阅创建启用了前图像和后图像的集合部分。

• 配置你的 change stream 以检索前映像和后映像中的一个或两个。

  提示

  要配置变更流以记录变更事件中的前像，请参阅前像配置示例。

  要配置变更流以记录变更事件中的后映像，请参阅后映像配置示例。

CreateCollectionOptions collectionOptions = new CreateCollectionOptions();
collectionOptions.changeStreamPreAndPostImagesOptions(new ChangeStreamPreAndPostImagesOptions(true));
database.createCollection("myColl", collectionOptions);

MongoCollection<Document> collection = database.getCollection("myColl");
ChangeStreamIterable<Document> changeStream = collection.watch()
        .fullDocumentBeforeChange(FullDocumentBeforeChange.REQUIRED);
changeStream.forEach(event ->
        System.out.println("Received a change: " + event));

Received a change: ChangeStreamDocument{
   operationType=update,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   destinationNamespace=null,
   fullDocument=null,
   fullDocumentBeforeChange=Document{{_id=..., amount=150, ...}},
   ...
}

MongoCollection<Document> collection = database.getCollection("myColl");
ChangeStreamIterable<Document> changeStream = collection.watch()
        .fullDocument(FullDocument.WHEN_AVAILABLE);
changeStream.forEach(event ->
        System.out.println("Received a change: " + event));

Received a change: ChangeStreamDocument{
   operationType=update,
   resumeToken={"_data": "..."},
   namespace=myDb.myColl,
   destinationNamespace=null,
   fullDocument=Document{{_id=..., color=purple, ...}},
   updatedFields={"color": purple},
   ...
}