Event-driven programmation avec Amazon DocumentDB et Java

Event-driven la programmation dans le contexte d'Amazon DocumentDB représente un modèle architectural puissant dans lequel les modifications apportées à la base de données sont les principaux générateurs d'événements qui déclenchent la logique et les processus métier ultérieurs. Lorsque des enregistrements sont insérés, mis à jour ou supprimés dans une collection DocumentDB, ces modifications agissent comme des événements qui lancent automatiquement divers processus, notifications ou tâches de synchronisation des données en aval. Ce modèle est particulièrement utile dans les systèmes distribués modernes où de multiples applications ou services doivent réagir aux changements de données en temps réel. Les flux de modifications constituent le principal mécanisme de mise en œuvre de la programmation pilotée par les événements dans DocumentDB.

Utilisation des flux de modifications depuis l'application Java

La watch() méthode du pilote Java de MongoDB est le principal mécanisme de surveillance des modifications de données en temps réel dans Amazon DocumentDB. La watch() méthode peut être appelée par des MongoCollection objets MongoClient MongoDatabase, et.

La watch() méthode renvoie une instance ChangeStreamIterable qui prend en charge diverses options de configuration, notamment la recherche complète de documents pour les mises à jour, la fourniture de jetons de CV et d'horodatage pour plus de fiabilité, et des étapes d'agrégation de pipeline pour filtrer les modifications.

ChangeStreamIterableimplémente l'interface Java principale Iterable et peut être utilisé avecforEach(). Pour capturer les événements en utilisantforEach(), transmettez une fonction de rappel à forEach() qui traite l'événement modifié. L'extrait de code suivant montre comment ouvrir un flux de modification sur une collection pour démarrer la surveillance des événements de changement :

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

Une autre façon de parcourir tous les événements de changement consiste à ouvrir un curseur qui maintient une connexion au cluster et reçoit en permanence les nouveaux événements de changement au fur et à mesure qu'ils se produisent. Pour obtenir un curseur de flux de modifications, utilisez la cursor() méthode ChangeStreamIterable object. L'exemple de code suivant montre comment surveiller les événements de changement à l'aide du curseur :

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

Il est recommandé de créer l'instruction MongoChangeStreamCursor in a try-with-resource ou de fermer manuellement le curseur. L'appel de la cursor() méthode on ChangeStreamIterable renvoie une MongoChangeStreamCursor valeur créée sur un ChangeStreamDocument objet.

La ChangeStreamDocument classe est un élément crucial qui représente les événements de changement individuels dans le flux. Il contient des informations détaillées sur chaque modification, notamment le type d'opération (insérer, mettre à jour, supprimer, remplacer), la clé du document, les informations relatives à l'espace de noms et le contenu complet du document lorsqu'il est disponible. La classe fournit des méthodes permettant d'accéder à divers aspects de l'événement de changement, tels que getOperationType() la détermination du type de modification, l'accès getFullDocument() à l'état complet du document et getDocumentKey() l'identification du document modifié.

L'ChangeStreamDocumentobjet fournit deux informations importantes, un jeton de reprise et l'heure de l'événement de changement.

Les jetons de reprise et les opérations temporelles dans les flux de modifications de DocumentDB fournissent des mécanismes essentiels pour maintenir la continuité et gérer l'accès aux données historiques. Un jeton de reprise est un identifiant unique généré pour chaque événement de changement. Il sert de signet et permet aux applications de relancer le traitement du flux de modifications à partir d'un point spécifique après des déconnexions ou des échecs. Lorsqu'un curseur de flux de modifications est créé, il peut utiliser un jeton de reprise précédemment stocké via resumeAfter() cette option, ce qui permet au flux de reprendre là où il s'est arrêté plutôt que de recommencer depuis le début ou de perdre des événements.

Time-based les opérations relatives aux flux de changement proposent différentes approches pour gérer le point de départ de la surveillance des événements de changement. L'startAtOperationTime()option vous permet de commencer à observer les modifications survenues au moment ou après un horodatage spécifique. Ces fonctionnalités temporelles sont particulièrement utiles dans les scénarios nécessitant le traitement des données historiques, la restauration instantanée ou la synchronisation entre les systèmes.

L'exemple de code suivant extrait l'événement associé au document d'insertion, capture son jeton de reprise, puis fournit ce jeton pour commencer à surveiller les événements après l'événement d'insertion. L'événement est associé à l'événement de mise à jour, puis obtient l'heure du cluster à laquelle la mise à jour a eu lieu et utilise cet horodatage comme point de départ pour un traitement ultérieur.

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

Par défaut, l'événement de modification de mise à jour n'inclut pas le document complet et inclut uniquement les modifications apportées. Si vous devez accéder au document complet qui a été mis à jour, vous pouvez appeler la fullDocument() méthode sur l'ChangeStreamIterableobjet. N'oubliez pas que lorsque vous demandez le renvoi d'un document complet pour un événement de mise à jour, le document qui existe au moment de l'appel pour modifier les flux est renvoyé.

Cette méthode prend une FullDocument énumération comme paramètre. Actuellement, Amazon DocumentDB ne prend en charge que DEFAULT et les UPDATE_LOOKUP valeurs. L'extrait de code suivant montre comment demander le document complet pour les événements de mise à jour lorsque vous commencez à surveiller les modifications :

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