Event-driven programmazione con Amazon DocumentDB e Java

Event-driven la programmazione nel contesto di Amazon DocumentDB rappresenta un potente modello architetturale in cui le modifiche al database fungono da generatori di eventi principali che attivano la logica e i processi aziendali successivi. Quando i record vengono inseriti, aggiornati o eliminati in una raccolta DocumentDB, queste modifiche agiscono come eventi che avviano automaticamente vari processi, notifiche o attività di sincronizzazione dei dati a valle. Questo modello è particolarmente utile nei moderni sistemi distribuiti in cui più applicazioni o servizi devono reagire alle modifiche dei dati in tempo reale. Il meccanismo principale per implementare la programmazione basata sugli eventi in DocumentDB è costituito dai flussi di modifiche.

Utilizzo dei flussi di modifica dall'applicazione Java

Il watch() metodo del driver Java di MongoDB è il meccanismo principale per monitorare le modifiche ai dati in tempo reale in Amazon DocumentDB. Il watch() metodo può essere richiamato da MongoClient MongoDatabase, e oggetti. MongoCollection

Il watch() metodo restituisce un'istanza ChangeStreamIterable che supporta varie opzioni di configurazione, tra cui la ricerca completa del documento per gli aggiornamenti, la fornitura di token di curriculum e timestamp per l'affidabilità e le fasi di aggregazione della pipeline per filtrare le modifiche.

ChangeStreamIterableimplementa l'interfaccia Java principale e può essere utilizzato con. Iterable forEach() Per catturare gli eventi utilizzandoforEach(), passate una funzione di callback forEach() che elabori l'evento modificato. Il seguente frammento di codice mostra come aprire un flusso di modifica su una raccolta per avviare il monitoraggio degli eventi di modifica:

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

Un altro modo per visualizzare tutti gli eventi di modifica consiste nell'aprire un cursore che mantiene una connessione al cluster e riceve continuamente nuovi eventi di modifica non appena si verificano. Per ottenere un cursore dei flussi di modifica, utilizzate il metodo dell'oggetto. cursor() ChangeStreamIterable Il seguente esempio di codice mostra come monitorare gli eventi di modifica utilizzando il cursore:

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

Come procedura consigliata, è possibile creare l'istruzione MongoChangeStreamCursor in a try-with-resource o chiudere manualmente il cursore. La chiamata al cursor() metodo on ChangeStreamIterable restituisce un elemento creato su un MongoChangeStreamCursor oggetto. ChangeStreamDocument

La ChangeStreamDocument classe è un componente cruciale che rappresenta i singoli eventi di modifica nello stream. Contiene informazioni dettagliate su ciascuna modifica, tra cui il tipo di operazione (inserimento, aggiornamento, eliminazione, sostituzione), la chiave del documento, le informazioni sullo spazio dei nomi e il contenuto completo del documento, se disponibile. La classe fornisce metodi per accedere a vari aspetti dell'evento di modifica, getOperationType() ad esempio per determinare il tipo di modifica, accedere getFullDocument() allo stato completo del documento e getDocumentKey() identificare il documento modificato.

L'ChangeStreamDocumentoggetto fornisce due informazioni importanti, un token di curriculum e l'ora dell'evento di modifica.

I token di curriculum e le operazioni basate sul tempo nei flussi di modifica di DocumentDB forniscono meccanismi cruciali per mantenere la continuità e gestire la cronologia degli accessi ai dati. Un token di curriculum è un identificatore univoco generato per ogni evento di modifica, che funge da segnalibro che consente alle applicazioni di riavviare l'elaborazione del flusso di modifiche da un punto specifico dopo disconnessioni o errori. Quando viene creato un cursore del flusso di modifiche, può utilizzare un token di curriculum precedentemente memorizzato tramite l'resumeAfter()opzione, consentendo allo stream di continuare dal punto in cui era stato interrotto anziché ricominciare dall'inizio o perdere eventi.

Time-based le operazioni nei flussi di modifica offrono approcci diversi per gestire il punto di partenza del monitoraggio degli eventi di modifica. L'startAtOperationTime()opzione consente di iniziare a guardare le modifiche avvenute in corrispondenza o dopo un determinato timestamp. Queste funzionalità basate sul tempo sono particolarmente utili in scenari che richiedono l'elaborazione di dati storici, il ripristino point-in-time o la sincronizzazione tra sistemi.

L'esempio di codice seguente recupera l'evento associato al documento di inserimento, ne acquisisce il token di curriculum e quindi fornisce quel token per avviare il monitoraggio degli eventi successivi all'evento di inserimento. L'evento è associato all'evento di aggiornamento, quindi ottiene l'ora del cluster in cui è avvenuto l'aggiornamento e utilizza tale timestamp come punto di partenza per ulteriori elaborazioni.

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

Per impostazione predefinita, l'evento di modifica dell'aggiornamento non include il documento completo e include solo le modifiche apportate. Se è necessario accedere al documento completo che è stato aggiornato, è possibile chiamare il fullDocument() metodo sull'ChangeStreamIterableoggetto. Tieni presente che quando richiedi la restituzione di un documento completo per un evento di aggiornamento, viene restituito il documento esistente nel momento in cui viene effettuata la chiamata per modificare i flussi di modifica.

Questo metodo accetta un FullDocument enum come parametro. Attualmente, Amazon DocumentDB supporta solo DEFAULT e UPDATE_LOOKUP valori. Il seguente frammento di codice mostra come richiedere il documento completo per gli eventi di aggiornamento quando si inizia a controllare le modifiche:

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