オープンChange Streams

Overview

このガイドでは、変更ストリームを使用してデータベースへのリアルタイムの変更を監視する方法を学習できます。 変更ストリームは、アプリケーションが単一のコレクション、データベース、または配置のデータ変更をサブスクライブできる MongoDB Server の機能です。

アプリケーションが受信したデータをフィルタリングして変換するための集計演算子のセットを指定できます。 MongoDB 配置 v6.0 以降に接続する場合は、変更前と変更後のドキュメント データを含めるようにイベントを構成することもできます。

変更ストリームを開いて構成する方法については、次のセクションを参照してください。

• 変更ストリームを開く

• 変更ストリームへの集計演算子の適用

• 変更前と変更後のイメージを含みます

変更ストリームを開く

変更ストリームを開いて、特定のタイプのデータ変更をサブスクライブし、アプリケーション内で変更イベントを生成できます。

変更ストリームを開くには、 MongoCollection 、 MongoDatabase 、またはMongoClientのインスタンスで watch()メソッドを呼び出します。

watch()メソッドを呼び出すオブジェクトによって、変更ストリームがリッスンするイベントの範囲が決まります。

MongoCollectionでwatch()を呼び出すと、変更ストリームによってコレクションが監視されます。

MongoDatabaseでwatch()を呼び出すと、変更ストリームによってそのデータベース内のすべてのコレクションが監視されます。

MongoClientでwatch()を呼び出すと、変更ストリームによって接続された 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,
   ...
}

変更ストリームへの集計演算子の適用

集計パイプラインをパラメーターとしてwatch()メソッドに渡して、変更ストリームが受信する変更イベントを指定できます。

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,
   ...
}

大規模な変更ストリーム イベントの分裂

MongoDB 7.0 以降では、 $changeStreamSplitLargeEvent集計ステージを使用して 16 MB を超えるイベントを小さなフラグメントに分割できます。

厳密に必要な場合にのみ$changeStreamSplitLargeEventを使用してください。 たとえば、アプリケーションで完全なドキュメントの変更前または変更後のイメージが必要で、16 MB を超えるイベントが生成される場合は、 $changeStreamSplitLargeEventを使用します。

$changeStreamSplitLargeEvent ステージはフラグメントを順番に返します。 変更ストリーム カーソル を使用してフラグメントにアクセスできます。 各フラグメントには、次のフィールドを含むSplitEventオブジェクトが含まれます。

次の例では、 $changeStreamSplitLargeEvent集計ステージを使用して大規模なイベントを分割し、変更ストリームを変更します。

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

次の例に示すように、変更ストリーム カーソルでgetSplitEvent()メソッドを呼び出してSplitEventにアクセスできます。

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

$changeStreamSplitLargeEvent集計ステージの詳細については、 $changeStreamSplitLargetサーバーのドキュメントを参照してください。

変更前と変更後のイメージを含みます

変更イベントは、以下のデータを含めるか省略するように構成できます。

• 変更前のイメージ: 操作前のドキュメントのバージョンを表すドキュメント（存在する場合）

• 変更後のイメージ: 操作後のドキュメントのバージョンを表すドキュメント（存在する場合）

変更前イメージまたは変更後イメージを含む変更ストリーム イベントを受信するには、次のアクションを実行する必要があります。

• MongoDB 配置のコレクションの変更前と変更後のイメージを有効にします。

  Tip

  配置で変更前と変更後のイメージを有効にする方法については、サーバー マニュアルの「 Change Streams とドキュメントの変更前イメージおよび変更後イメージ 」を参照してください。

  変更前と変更後のイメージが有効になっているコレクションを作成するようにドライバーに指示する方法については、「 変更前と変更後のイメージが有効になっているコレクションの作成 」セクションを参照してください。

• 変更ストリームを設定して、変更前のイメージと変更後のイメージのどちらかを取得します。

  Tip

  変更イベントに変更前のイメージを記録するように変更ストリームを構成する方法について詳しくは、「 変更前イメージの構成例 」を参照してください。

  変更イベントで変更後のイメージを記録するように変更ストリームを構成するには、「 変更後のイメージの構成例 」を参照してください。

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},
   ...
}