View a markdown version of this page

$ Stream ändern - Amazon DocumentDB
Beispiel (MongoDB Shell)Codebeispiele

Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

$ Stream ändern

Wird von Elastic Cluster nicht unterstützt.

In der $changeStream Aggregationsphase wird ein Change-Stream-Cursor geöffnet, um Änderungen an einer Sammlung in Echtzeit zu überwachen. Sie gibt Änderungsereignisdokumente zurück, wenn Einfüge-, Aktualisierungs-, Ersetzungs- oder Löschvorgänge ausgeführt werden.

Parameter

  • fullDocument: Gibt an, ob bei Aktualisierungsvorgängen das gesamte Dokument zurückgegeben werden soll. Die Optionen lauten default oder updateLookup.

  • resumeAfter: Optional. Setzen Sie das Token fort, um von einem bestimmten Punkt im Change-Stream aus fortzufahren.

  • startAtOperationTime: Optional. Zeitstempel, ab dem der Change-Stream gestartet werden soll.

  • allChangesForCluster: Optional. Boolescher Wert: Wanntrue, überwacht alle Änderungen im Cluster (für die Admin-Datenbank). Wenn false (Standardeinstellung), überwacht nur die angegebene Sammlung.

  • showExpandedEvents: Optional. Boolescher Wert: Wenntrue, schließt DDL-Ereignisbenachrichtigungen (Expanded Data Definition Language) wie createCollection und createIndexes in den Change-Stream ein. Wenn false (Standardeinstellung), werden diese Ereignisse nicht zurückgegeben. Verfügbar in Amazon DocumentDB 8.0 ab Engine-Version 8.0.2.

Beispiel (MongoDB Shell)

Das folgende Beispiel zeigt, wie die $changeStream Bühne verwendet wird, um Änderungen an einer Sammlung zu überwachen.

Beispiel für eine Abfrage 

// 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()));
}

Ausgabe

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

Codebeispiele

Um ein Codebeispiel für die Verwendung der $changeStream Aggregationsstufe anzuzeigen, wählen Sie die Registerkarte für die Sprache, die Sie verwenden möchten:

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()