Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Modèle de boîte d’envoi transactionnelle

Intention

Le modèle de boîte d’envoi transactionnelle résout le problème des opérations d’écriture double qui se produit dans les systèmes distribués lorsqu’une seule opération implique à la fois une opération d’écriture dans la base de données et une notification de message ou d’événement. Une opération d’écriture double se produit lorsqu’une application écrit sur deux systèmes différents. Par exemple, lorsqu’un microservice doit conserver des données dans la base de données et envoyer un message pour avertir les autres systèmes. L’échec de l’une de ces opérations peut entraîner des données incohérentes.

Motivation

Lorsqu’un microservice envoie une notification d’événement après une mise à jour de base de données, ces deux opérations doivent être exécutées de manière atomique pour garantir la fiabilité et la cohérence des données.

Applicabilité

Utilisez le modèle de boîte d’envoi transactionnelle lorsque :

Problèmes et considérations

Mise en œuvre

Architecture de haut niveau

Le diagramme de séquence suivant montre l’ordre des événements qui se produisent lors des opérations d’écriture double.

Si la mise à jour de la base de données des vols échoue, mais que la notification est envoyée, le service de paiement traitera le paiement en fonction de la notification d’événement. Cela entraînera des incohérences dans les données en aval.

Mise en œuvre à l’aide des services AWS

Pour illustrer le schéma dans le diagramme de séquence, nous utiliserons les AWS services suivants, comme indiqué dans le schéma suivant.

Si le service de vol échoue après avoir validé la transaction, la notification d’événement peut ne pas être envoyée.

Cependant, la transaction peut échouer et être restaurée, mais la notification d’événement peut tout de même être envoyée, obligeant le service de paiement à traiter le paiement.

Pour résoudre ce problème, vous pouvez utiliser une table de boîte d’envoi ou la capture des données de modification (CDC). Les sections suivantes traitent de ces deux options et de la manière dont vous pouvez les mettre en œuvre à l’aide des services AWS.

Utilisation d’une table de boîte d’envoi avec une base de données relationnelle

Une table de boîte d’envoi stocke tous les événements du service de vol avec un horodatage et un numéro de séquence.

Lorsque la table de vol est mise à jour, la table de boîte d’envoi est également mise à jour dans le cadre de la même transaction. Un autre service (par exemple, le service de traitement des événements) lit la table de la boîte d’envoi et envoie l’événement à Amazon SQS. Amazon SQS envoie un message concernant l’événement au service de paiement pour un traitement ultérieur. Les files d’attente standard Amazon SQS garantissent que le message est délivré au moins une fois et qu’il ne soit pas perdu. Toutefois, lorsque vous utilisez des files d’attente standard Amazon SQS, le même message ou événement peut être transmis plusieurs fois. Vous devez donc vous assurer que le service de notification d’événements est idempotent (c’est-à-dire que le traitement du même message plusieurs fois ne devrait pas avoir d’effet négatif). Si vous souhaitez que le message soit livré une seule fois, vous pouvez utiliser les files d’attente FIFO (premier entré, premier sorti) d’Amazon SQS dans le cadre de l’ordre des messages.

Si la mise à jour de la table de vol ou la mise à jour de la table de boîte d’envoi échouent, l’intégralité de la transaction est restaurée, de sorte qu’il n’y a aucune incohérence dans les données en aval.

Dans le schéma suivant, l’architecture de boîte d’envoi transactionnelle est mise en œuvre à l’aide d’une base de données Amazon RDS. Lorsque le service de traitement des événements lit la table de la boîte d’envoi, il reconnaît uniquement les lignes faisant partie d’une transaction validée (réussie), puis place le message correspondant à l’événement dans la file d’attente SQS, qui est lue par le service de paiement pour un traitement ultérieur. Cette conception résout le problème des opérations d’écriture double et préserve l’ordre des messages et des événements en utilisant des horodatages et des numéros de séquence.

Utilisation de la capture des données de modification (CDC)

Certaines bases de données prennent en charge la publication de modifications au niveau des éléments afin de capturer les données modifiées. Vous pouvez identifier les éléments modifiés et envoyer une notification d’événement en conséquence. Cela permet d’économiser les frais liés à la création d’une autre table pour suivre les mises à jour. L’événement initié par le service de vol est enregistré dans un autre attribut du même élément.

Amazon DynamoDB est une base de données NoSQL clé-valeur qui prend en charge les mises à jour CDC. Dans le diagramme de séquence suivant, DynamoDB publie les modifications apportées au niveau des éléments dans Amazon DynamoDB Streams. Le service de traitement des événements lit les flux et publie la notification d’événement dans le service de paiement pour un traitement ultérieur.

DynamoDB Streams capture le flux d’informations relatives aux modifications apportées au niveau des éléments dans une table DynamoDB à l’aide d’une séquence temporelle.

Vous pouvez implémenter un modèle de boîte d’envoi transactionnelle en activant les flux dans la table DynamoDB. La fonction Lambda du service de traitement des événements est associée à ces flux.

Exemple de code

Utilisation d'une table de boîte d'envoi

L'exemple de code présenté dans cette section montre comment implémenter le modèle de boîte d'envoi transactionnelle à l'aide d'une table de boîte d'envoi. Pour consulter le code complet, consultez le GitHubréférentiel de cet exemple.

L’extrait de code suivant enregistre l’entité Flight et l’événement Flight dans la base de données dans leurs tables respectives au cours d’une seule transaction.

@PostMapping("/flights")
    @Transactional
    public Flight createFlight(@Valid @RequestBody Flight flight) {
        Flight savedFlight = flightRepository.save(flight);
        JsonNode flightPayload = objectMapper.convertValue(flight, JsonNode.class);
        FlightOutbox outboxEvent = new FlightOutbox(flight.getId().toString(), FlightOutbox.EventType.FLIGHT_BOOKED,
                flightPayload);
        outboxRepository.save(outboxEvent);
        return savedFlight;
    }

Un service distinct est chargé d’analyser régulièrement la table de la boîte d’envoi pour détecter de nouveaux événements, de les envoyer à Amazon SQS et de les supprimer de la table si Amazon SQS répond correctement. Le taux d’interrogation est configurable dans le fichier application.properties.

@Scheduled(fixedDelayString = "${sqs.polling_ms}")
    public void forwardEventsToSQS() {
        List<FlightOutbox> entities = outboxRepository.findAllByOrderByIdAsc(Pageable.ofSize(batchSize)).toList();
        if (!entities.isEmpty()) {
            GetQueueUrlRequest getQueueRequest = GetQueueUrlRequest.builder()
                    .queueName(sqsQueueName)
                    .build();
            String queueUrl = this.sqsClient.getQueueUrl(getQueueRequest).queueUrl();
            List<SendMessageBatchRequestEntry> messageEntries = new ArrayList<>();
            entities.forEach(entity -> messageEntries.add(SendMessageBatchRequestEntry.builder()
                    .id(entity.getId().toString())
                    .messageGroupId(entity.getAggregateId())
                    .messageDeduplicationId(entity.getId().toString())
                    .messageBody(entity.getPayload().toString())
                    .build())
            );
            SendMessageBatchRequest sendMessageBatchRequest = SendMessageBatchRequest.builder()
                    .queueUrl(queueUrl)
                    .entries(messageEntries)
                    .build();
            sqsClient.sendMessageBatch(sendMessageBatchRequest);
            outboxRepository.deleteAllInBatch(entities);
        }
    }

Utilisation de la capture des données de modification (CDC)

L'exemple de code présenté dans cette section montre comment implémenter le modèle de boîte d'envoi transactionnelle en utilisant les fonctionnalités de capture des données de modification (CDC) de DynamoDB. Pour consulter le code complet, consultez le GitHubréférentiel de cet exemple.

L'extrait de AWS Cloud Development Kit (AWS CDK) code suivant crée une table de vol DynamoDB et un flux de données Amazon Kinesis (cdcStream), et configure la table de vol pour envoyer toutes ses mises à jour au flux.

Const cdcStream = new kinesis.Stream(this, 'flightsCDCStream', {
    streamName: 'flightsCDCStream'
})

const flightTable = new dynamodb.Table(this, 'flight', {
    tableName: 'flight',
    kinesisStream: cdcStream,
    partitionKey: {
        name: 'id',
        type: dynamodb.AttributeType.STRING,
    }

});

L'extrait de code et la configuration suivants définissent une fonction Spring Cloud Stream qui récupère les mises à jour dans le flux Kinesis et transmet ces événements à une file d'attente SQS pour un traitement ultérieur.

applications.properties
spring.cloud.stream.bindings.sendToSQS-in-0.destination=${kinesisstreamname}
spring.cloud.stream.bindings.sendToSQS-in-0.content-type=application/ddb
    
QueueService.java
@Bean
public Consumer<Flight> sendToSQS() {
    return this::forwardEventsToSQS;
}

public void forwardEventsToSQS(Flight flight) {
    GetQueueUrlRequest getQueueRequest = GetQueueUrlRequest.builder()
            .queueName(sqsQueueName)
            .build();
    String queueUrl = this.sqsClient.getQueueUrl(getQueueRequest).queueUrl();
    try {
        SendMessageRequest send_msg_request = SendMessageRequest.builder()
                .queueUrl(queueUrl)
                .messageBody(objectMapper.writeValueAsString(flight))
                .messageGroupId("1")
                .messageDeduplicationId(flight.getId().toString())
                .build();
        sqsClient.sendMessage(send_msg_request);
    } catch (IOException | AmazonServiceException e) {
        logger.error("Error sending message to SQS", e);
    }
}

GitHub référentiel

Pour une implémentation complète de l'exemple d'architecture pour ce modèle, consultez le GitHub référentiel à l'adresse https://github.com/aws-samples/ transactional-outbox-pattern.