変更ストリーム

このバージョンのドキュメントはアーカイブされており、サポートされなくなりました。5.0 配置をアップグレードするには、 MongoDB 6.0 のアップグレード手順を参照してください。

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

可用性

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

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

Stable API でのサポート

変更ストリームは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 から実行する必要があります。

次の例では、コレクションの変更ストリームを開き、カーソルを反復処理して変更ストリームドキュメントを検索します。[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();

以下の例では 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
  })
  .once('error', () => {
    // handle error
  });

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

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

以下の例では、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 の時間範囲内である必要があります。