$changeStream

弹性集群不支持。

$changeStream聚合阶段打开变更流光标以监控集合的实时更改。当插入、更新、替换或删除操作发生时，它会返回更改事件文档。

参数

• fullDocument：指定是否为更新操作返回完整文档。选项有 default 或 updateLookup。

• resumeAfter：可选。恢复令牌以从更改流中的特定点继续操作。

• startAtOperationTime：可选。启动变更流的时间戳。

• allChangesForCluster：可选。布尔值。何时true，监视集群中的所有更改（适用于管理员数据库）。当false（默认）时，仅监视指定的集合。

• showExpandedEvents：可选。布尔值。何时true，在更改流中包括扩展数据定义语言 (DDL) 事件通知createIndexes，例如createCollection和。当false（默认）时，不返回这些事件。从引擎版本 8.0.2 开始，在 Amazon DocumentDB 8.0 中可用。

示例（MongoDB Shell）

以下示例演示使用$changeStream舞台监控集合的更改。

查询示例

// Open change stream first
const changeStream = db.inventory.aggregate([
  { $changeStream: { fullDocument: "updateLookup" } }
]);

// In another session, insert a document
db.inventory.insertOne({ _id: 1, item: "Widget", qty: 10 });

// Back in the first session, read the change event
if (changeStream.hasNext()) {
  print(tojson(changeStream.next()));
}

输出

{
  _id: { _data: '...' },
  operationType: 'insert',
  clusterTime: Timestamp(1, 1234567890),
  fullDocument: { _id: 1, item: 'Widget', qty: 10 },
  ns: { db: 'test', coll: 'inventory' },
  documentKey: { _id: 1 }
}

代码示例

要查看使用$changeStream聚合阶段的代码示例，请选择要使用的语言的选项卡：

Node.js

const { MongoClient } = require('mongodb');

async function example() {
  const client = await MongoClient.connect('mongodb://<username>:<password>@<cluster-endpoint>:27017/?tls=true&tlsCAFile=global-bundle.pem&replicaSet=rs0&readPreference=secondaryPreferred&retryWrites=false');
  const db = client.db('test');
  const collection = db.collection('inventory');

  // Open change stream
  const changeStream = collection.watch([]);

  changeStream.on('change', (change) => {
    console.log('Change detected:', change);
  });

  // Simulate insert in another operation
  setTimeout(async () => {
    await collection.insertOne({ _id: 1, item: 'Widget', qty: 10 });
  }, 1000);

  // Keep connection open to receive changes
  // In production, handle cleanup appropriately
}

example();

Python

from pymongo import MongoClient
import threading
import time

def example():
    client = MongoClient('mongodb://<username>:<password>@<cluster-endpoint>:27017/?tls=true&tlsCAFile=global-bundle.pem&replicaSet=rs0&readPreference=secondaryPreferred&retryWrites=false')
    db = client['test']
    collection = db['inventory']

    # Open change stream
    change_stream = collection.watch([])

    # Insert document in separate thread after delay
    def insert_doc():
        time.sleep(1)
        collection.insert_one({'_id': 1, 'item': 'Widget', 'qty': 10})

    threading.Thread(target=insert_doc).start()

    # Watch for changes
    for change in change_stream:
        print('Change detected:', change)
        break  # Exit after first change

    client.close()

example()