Event-driven Programmierung mit Amazon DocumentDB und Java

Event-driven Die Programmierung im Kontext von Amazon DocumentDB stellt ein leistungsstarkes Architekturmuster dar, bei dem Datenbankänderungen als primäre Ereignisgeneratoren dienen, die nachfolgende Geschäftslogik und Prozesse auslösen. Wenn Datensätze in eine DocumentDB-Sammlung eingefügt, aktualisiert oder gelöscht werden, wirken diese Änderungen als Ereignisse, die automatisch verschiedene nachgelagerte Prozesse, Benachrichtigungen oder Datensynchronisierungsaufgaben auslösen. Dieses Muster ist besonders nützlich in modernen verteilten Systemen, in denen mehrere Anwendungen oder Dienste in Echtzeit auf Datenänderungen reagieren müssen. Der primäre Mechanismus zur Implementierung ereignisgesteuerter Programmierung in DocumentDB sind Change Streams.

Arbeiten mit Change-Streams aus der Java-Anwendung

Die watch() Methode im Java-Treiber von MongoDB ist der primäre Mechanismus zur Überwachung von Datenänderungen in Amazon DocumentDB in Echtzeit. Die watch() Methode kann von Objekten MongoClient MongoDatabase, und MongoCollection aufgerufen werden.

Die watch() Methode gibt eine Instanz zurück ChangeStreamIterable, die verschiedene Konfigurationsoptionen unterstützt, darunter die vollständige Dokumentsuche nach Aktualisierungen, die Bereitstellung von Lebenslauf-Tokens und Zeitstempeln für die Zuverlässigkeit sowie Pipeline-Aggregationsstufen zum Filtern von Änderungen.

ChangeStreamIterableimplementiert die Java-Kernschnittstelle Iterable und kann mit verwendet werden. forEach() Um Ereignisse mit zu erfassenforEach(), übergeben Sie eine Callback-FunktionforEach(), die das geänderte Ereignis verarbeitet. Der folgende Codeausschnitt zeigt, wie ein Change-Stream für eine Sammlung geöffnet wird, um die Überwachung von Änderungsereignissen zu starten:

ChangeStreamIterable < Document > iterator = collection.watch();
iterator.forEach(event - > {
    System.out.println("Received a change: " + event);
});

Eine andere Möglichkeit, alle Änderungsereignisse zu durchlaufen, besteht darin, einen Cursor zu öffnen, der eine Verbindung zum Cluster aufrechterhält und kontinuierlich neue Änderungsereignisse empfängt, sobald sie auftreten. Um einen Change-Streams-Cursor zu erhalten, verwenden Sie die cursor() Methode object. ChangeStreamIterable Das folgende Codebeispiel zeigt, wie Änderungsereignisse mithilfe des Cursors überwacht werden:

try (MongoChangeStreamCursor < ChangeStreamDocument < Document >> cursor = collection.watch().cursor()) {
    System.out.println(cursor.tryNext());
}

Es hat sich bewährt, entweder die Anweisung MongoChangeStreamCursor in a try-with-resource zu erstellen oder den Cursor manuell zu schließen. Wenn Sie die cursor() Methode aufrufen, wird eine ChangeStreamIterable zurückgegebenMongoChangeStreamCursor, die über einem Objekt erstellt wurde. ChangeStreamDocument

Die ChangeStreamDocument Klasse ist eine wichtige Komponente, die einzelne Änderungsereignisse im Stream darstellt. Sie enthält detaillierte Informationen zu jeder Änderung, einschließlich des Vorgangstyps (Einfügen, Aktualisieren, Löschen, Ersetzen), des Dokumentschlüssels, der Namespace-Informationen und des vollständigen Dokumentinhalts, sofern verfügbar. Die Klasse bietet Methoden für den Zugriff auf verschiedene Aspekte des Änderungsereignisses, z. B. getOperationType() um die Art der Änderung getFullDocument() zu bestimmen, auf den vollständigen Dokumentstatus zuzugreifen und das geänderte Dokument getDocumentKey() zu identifizieren.

Das ChangeStreamDocument Objekt stellt zwei wichtige Informationen bereit: ein Resume-Token und die Uhrzeit des Änderungsereignisses.

Resume-Token und zeitbasierte Operationen in DocumentDB-Change-Streams bieten entscheidende Mechanismen zur Aufrechterhaltung der Kontinuität und zur Verwaltung des Zugriffs auf historische Daten. Ein Resume-Token ist eine eindeutige Kennung, die für jedes Änderungsereignis generiert wird. Sie dient als Lesezeichen, das es Anwendungen ermöglicht, die Change-Stream-Verarbeitung nach Verbindungsabbrüchen oder Ausfällen von einem bestimmten Punkt aus neu zu starten. Wenn ein Change-Stream-Cursor erstellt wird, kann er über die resumeAfter() Option ein zuvor gespeichertes Resume-Token verwenden, sodass der Stream an der Stelle fortgesetzt werden kann, an der er aufgehört hat, anstatt am Anfang zu beginnen oder Ereignisse zu verlieren.

Time-based Operationen in Change-Streams bieten unterschiedliche Ansätze, um den Ausgangspunkt der Überwachung von Änderungsereignissen zu verwalten. Mit dieser startAtOperationTime() Option können Sie beginnen, Änderungen zu beobachten, die zu oder nach einem bestimmten Zeitstempel eingetreten sind. Diese zeitbasierten Funktionen sind besonders nützlich in Szenarien, in denen historische Daten verarbeitet, zu einem bestimmten Zeitpunkt wiederhergestellt oder zwischen Systemen synchronisiert werden müssen.

Im folgenden Codebeispiel wird das mit dem Einfügedokument verknüpfte Ereignis abgerufen, das zugehörige Resume-Token erfasst und dann dieses Token bereitgestellt, um mit der Überwachung von Ereignissen nach dem Insert-Ereignis zu beginnen. Das Ereignis wird dem Aktualisierungsereignis zugeordnet, ruft dann die Cluster-Uhrzeit ab, zu der die Aktualisierung stattgefunden hat, und verwendet diesen Zeitstempel als Ausgangspunkt für die weitere Verarbeitung.

BsonDocument resumeToken;
BsonTimestamp resumeTime;

try (MongoChangeStreamCursor < ChangeStreamDocument < Document >> cursor = collection.watch().cursor()) {
    System.out.println("****************** Insert Document *******************");
    ChangeStreamDocument < Document > insertChange = cursor.tryNext();
    resumeToken = insertChange.getResumeToken();
    printJson(cursor.tryNext());
}
try (MongoChangeStreamCursor < ChangeStreamDocument < Document >> cursor = collection.watch()
    .resumeAfter(resumeToken)
    .cursor()) {
    System.out.println("****************** Update Document *******************");
    ChangeStreamDocument < Document > insertChange = cursor.tryNext();
    resumeTime = insertChange.getClusterTime();
    printJson(cursor.tryNext());
}
try (MongoChangeStreamCursor < ChangeStreamDocument < Document >> cursor = collection.watch()
    .startAtOperationTime(resumeTime)
    .cursor()) {
    System.out.println("****************** Delete Document *******************");
    printJson(cursor.tryNext());
  }

Standardmäßig umfasst das Aktualisierungsereignis nicht das gesamte Dokument und nur die vorgenommenen Änderungen. Wenn Sie auf das komplette Dokument zugreifen müssen, das aktualisiert wurde, können Sie die fullDocument() Methode für das ChangeStreamIterable Objekt aufrufen. Denken Sie daran, dass, wenn Sie die Rückgabe eines vollständigen Dokuments für ein Aktualisierungsereignis anfordern, das Dokument zurückgegeben wird, das zum Zeitpunkt des Aufrufs von Change Streams vorhanden ist.

Diese Methode verwendet eine FullDocument Aufzählung als Parameter. Derzeit unterstützt Amazon DocumentDB nur DEFAULT und UPDATE_LOOKUP Werte. Der folgende Codeausschnitt zeigt, wie Sie das vollständige Dokument für Aktualisierungsereignisse anfordern, wenn Sie beginnen, auf Änderungen zu achten:

try (MongoChangeStreamCursor < ChangeStreamDocument < Document >> cursor = collection.watch().fullDocument(FullDocument.UPDATE_LOOKUP).cursor())