変更ストリーム

項目一覧

可用性

接続
コレクション、データベース、または配置の監視
変更ストリームのパフォーマンスに関する考慮事項
変更ストリームを開く
変更ストリーム出力の修正
更新操作のための完全なドキュメントの検索
変更ストリームの再開
ユースケース
アクセス制御
イベント通知
照合
Change Streams と孤立したドキュメント
変更ストリームにおけるドキュメントの変更前と変更後のイメージ

変更ストリームを使用すると、 oplog を手動で追跡する以前の複雑さやリスクなしで、アプリケーションがリアルタイムデータ変更にアクセスできます。アプリケーションは変更ストリームを使用して、単一のコレクション、データベース、または配置全体のすべてのデータ変更にサブスクライブし、それらに即時に対応できます。変更ストリームは集計フレームワークを使用するため、アプリケーションは特定の変更をフィルタリングすることも、通知を任意に変換することもできます。

MongoDB 5.1 以降では、変更ストリームが最適化され、リソースの使用効率が上がり、一部の集計パイプラインステージの実行が高速化されています。

可用性

変更ストリームは、レプリカセットとシャーディングされたクラスターで使用できます。

ストレージエンジン
レプリカセットとシャーディングされたクラスターでは、WiredTiger ストレージエンジンを使用する必要があります。変更ストリームは、MongoDB の保管時の暗号化機能を採用した配置でも使用できます。
レプリカセットプロトコルバージョン
レプリカセットとシャーディングされたクラスターは、レプリカセットプロトコルバージョン 1（pv1）を使用する必要があります。
読み取り保証 (read concern) "過半数" 有効化
変更ストリームは、 "majority"の読み取り保証 (read concern) のサポートに関係なく使用できます。つまり、変更ストリームを使用するには、読み取り保証（read concern） majorityのサポートを有効にする（デフォルト）か無効にするかを選択できます。

Stable API でのサポート

変更ストリームはStable API V 1に含まれています。ただし、 showExpandedEventsオプションは Stable API V 1に含まれていません。

接続

変更ストリームの接続は、+srv 接続オプションを使用して DNS シードリスト、または接続文字列でサーバーを個別にリストする方法のいずれかを使用できます。

変更ストリームへの接続を失うか、接続が切断された場合、ドライバーは、読み込み設定 (read preference) が一致するクラスター内の別のノードを介して変更ストリームへの接続を再確立しようとします。ドライバーが正しく読み込み設定 (read preference) されているノードを見つけられない場合、例外が発生します。

詳細については、「接続文字列 URI 形式」を参照してください。

コレクション、データベース、または配置の監視

変更ストリームを以下に対して開くことができます。

ターゲット	説明
コレクション	単一のコレクション（`system` コレクション、または `admin` データベース、`local` データベース、`config` データベース内のコレクションを除く）の変更ストリームカーソルを開くことができます。このページの例では、MongoDB ドライバーを使用して、単一のコレクションの変更ストリームカーソルを開いて操作します。`mongosh` メソッド `db.collection.watch()` も参照してください。
データベース	単一データベース（`admin`、`local`、`config` データベースを除く）の変更ストリームカーソルを開いて、すべての非システムコレクションに対する変更を監視できます。 MongoDB ドライバーメソッドについては、「ドライバーのドキュメント」を参照してください。 `mongosh`メソッド`db.watch()`も参照してください。
配置	配置（レプリカセットまたはシャーディングされたクラスター）の変更ストリームカーソルを開き、すべてのデータベース（`admin`、`local`、`config` を除く）でシステム以外のコレクションに加えられた変更を監視できます。 MongoDB ドライバーメソッドについては、「ドライバーのドキュメント」を参照してください。 `mongosh`メソッド`Mongo.watch()`も参照してください。

注意

変更ストリームの例

このページの例では、MongoDB ドライバーを使用して、コレクションの変更ストリームカーソルを開き、変更ストリームカーソルを操作する方法を説明します。

変更ストリームのパフォーマンスに関する考慮事項

データベースに対して開かれたアクティブな変更ストリームの量が接続プールのサイズを超えると、通知のレイテンシが発生する可能性があります。各変更ストリームは、次のイベントを待機する間、変更ストリームに対する接続と getMore 操作を使用します。レイテンシの問題を回避するには、開かれた変更ストリームの数よりもプールサイズの方が大きいことを確認しておく必要があります。詳細については、maxPoolSize の設定を参照してください。

シャーディングされたクラスターの考慮事項

変更ストリームをシャーディングされたクラスター上で開くとき

mongos は、各シャードに個別の変更ストリームを作成します。この動作は、変更ストリームが特定のシャードキーの範囲を対象にしているかどうかに関わらず発生します。
mongos は変更ストリームの結果を受け取ると、その結果をソートおよびフィルタリングします。必要に応じて、mongos はfullDocument ルックアップも実行します。

最高のパフォーマンスを得るには、変更ストリームでの $lookup クエリの使用を制限します。

変更ストリームを開く

変更ストリームを開くには

レプリカセットの場合、データを保持しているいずれのノードからでも、変更ストリームを開く操作を実行できます。
シャーディングされたクラスターの場合、変更ストリームを開く操作は mongos から実行する必要があります。

次の例では、コレクションの変更ストリームを開き、カーソルを反復処理して変更ストリームドキュメントを検索します。[1]

➤ 右上の言語の選択のドロップダウンメニューを使用して、このページの例の言語を設定します。

以下のCの例ではMongoDBレプリカセットに接続し、 inventoryコレクションを含むデータベースにアクセスしていることを前提としています。

mongoc_collection_t *collection;
bson_t *pipeline = bson_new ();
bson_t opts = BSON_INITIALIZER;
mongoc_change_stream_t *stream;
const bson_t *change;
const bson_t *resume_token;
bson_error_t error;
collection = mongoc_database_get_collection (db, "inventory");
stream = mongoc_collection_watch (collection, pipeline, NULL /* opts */);
mongoc_change_stream_next (stream, &change);
if (mongoc_change_stream_error_document (stream, &error, NULL)) {
   MONGOC_ERROR ("%s\n", error.message);
}
mongoc_change_stream_destroy (stream);

以下のC#の例では、 MongoDBレプリカセットに接続し、 inventoryコレクションを含むデータベースにアクセスしていることを前提としています。

var cursor = inventory.Watch();
while (cursor.MoveNext() && cursor.Current.Count() == 0) { } // keep calling MoveNext until we've read the first batch
var next = cursor.Current.First();
cursor.Dispose();

以下の Go の例では、MongoDBレプリカセットに接続し、inventory コレクションを含むデータベースにアクセスしていることを前提としています。

cs, err := coll.Watch(ctx, mongo.Pipeline{})
assert.NoError(t, err)
defer cs.Close(ctx)
ok := cs.Next(ctx)
next := cs.Current

以下の Java の例では、MongoDB レプリカセットに接続し、inventory コレクションを含むデータベースにアクセスしていることを前提としています。

MongoCursor<ChangeStreamDocument<Document>> cursor = inventory.watch().iterator();
ChangeStreamDocument<Document> next = cursor.next();

以下の Kotlin の例では、 MongoDB レプリカセットに接続し、かつinventoryコレクションを含むデータベースにアクセスできることを前提としています。これらのタスクの完了の詳細については、「 Kotlin コルーチンドライバーのデータベースとコレクション」のガイドを参照してください。

val job = launch {
    val changeStream = collection.watch()
    changeStream.collect {
        println("Received a change event: $it")
    }
}

以下の例では MongoDB レプリカセットに接続し、アクセスしているデータベースに inventory コレクションが含まれていることを前提としています。

cursor = db.inventory.watch()
document = await cursor.next()

以下の Node.js の例では、MongoDB レプリカセットに接続し、アクセスしているデータベースに inventory コレクションが含まれていることを前提としています。

次の例では、ストリームを使用して変更イベントを処理しています。

const collection = db.collection('inventory');
const changeStream = collection.watch();
changeStream.on('change', next => {
  // process next document
});

代わりに、イテレーターを使用して変更イベントを処理することもできます。

const collection = db.collection('inventory');
const changeStream = collection.watch();
const next = await changeStream.next();

changeStream により EventEmitter が拡張されます。

以下の例では、MongoDB レプリカセットに接続し、 inventory コレクションを含むデータベースにアクセスしていることを前提としています。

$changeStream = $db->inventory->watch();
$changeStream->rewind();
$firstChange = $changeStream->current();
$changeStream->next();
$secondChange = $changeStream->current();

以下の Python の例では、 MongoDB レプリカセットに接続し、かつinventoryコレクションを含むデータベースにアクセスしていることを前提としています。

cursor = db.inventory.watch()
next(cursor)

以下の例では、MongoDB レプリカセットに接続し、 inventory コレクションを含むデータベースにアクセスしていることを前提としています。

cursor = inventory.watch.to_enum
next_change = cursor.next

以下の Swift（Async）例では MongoDB レプリカセットに接続し、データベースにアクセスしているinventory ことを前提としています。これにはコレクションが含まれます。

let inventory = db.collection("inventory")
// Option 1: retrieve next document via next()
let next = inventory.watch().flatMap { cursor in
    cursor.next()
}
// Option 2: register a callback to execute for each document
let result = inventory.watch().flatMap { cursor in
    cursor.forEach { event in
        // process event
        print(event)
    }
}

以下の Swift（Sync）例では MongoDB レプリカセットに接続し、アクセスしているデータベースに inventory コレクションが含まれていることを前提としています。

let inventory = db.collection("inventory")
let changeStream = try inventory.watch()
let next = changeStream.next()

カーソルからデータ変更イベントを検索するには、変更ストリームのカーソルを反復処理します。変更ストリームイベントの詳細については、「変更イベント」を参照してください。

変更ストリームカーソルは、次のいずれかが発生するまで開いたままになります。

カーソルが明示的に閉じている。
無効化イベントが発生している（コレクションの削除や名前の変更など）。
MongoDB 配置への接続が閉じているか、タイムアウトになっている。詳細については、「カーソルの動作」を参照してください。
配置がシャーディングされたクラスターの場合、シャードを削除すると、開いている変更ストリームのカーソルが閉じることがあります。閉じた変更ストリームのカーソルは完全に再開できない場合があります。

注意

閉じられていないカーソルのライフサイクルは言語に依存します。

[1]	`startAtOperationTime` を指定すると、特定の時点でカーソルを開くことができます。過去の開始点を指定する場合、oplog の時間範囲内である必要があります。