Messagerie FIFO avec Azure Service Bus et Spring Boot

Introduction

Azure Service Bus est un courtier de messages entièrement géré qui facilite une communication fiable entre les applications distribuées. Pour les applications qui nécessitent que les messages soient traités dans un ordre spécifique, par exemple pour garantir une séquence premier entré, premier sorti (FIFO), les sessions dans Azure Service Bus fournissent un mécanisme efficace pour le traitement des messages.

Dans le contexte d'une application Spring Boot, l'exploitation des sessions sur un sujet Azure Service Bus garantit que les messages avec le même ID de session sont traités dans le bon ordre par un seul consommateur à la fois. Cette solution est particulièrement utile lors de la gestion de scénarios de messagerie à haut débit tout en conservant l'ordre des messages.

Ce guide explique comment configurer une application Spring Boot pour consommer les messages d'Azure Service Bus dans l'ordre FIFO, garantissant ainsi la fiabilité et l'évolutivité en utilisant des sessions sans avoir besoin d'une infrastructure complexe.

En pratique

Pour qu'une application Spring Boot déployée sur plusieurs instances consomme les messages d'un sujet Azure Service Bus dans l'ordre FIFO, vous pouvez utiliser des sessions sur le sujet et configurer l'application pour gérer les sessions de manière coordonnée entre les autorités. Voici comment procéder :

1. Activer les sessions sur le sujet Azure Service Bus

• Assurez-vous que le sujet et l'abonnement Azure Service Bus sont configurés pour utiliser les sessions.
• Lors de l'envoi de messages, définissez un ID de session sur chaque message pour vous assurer que tous les messages avec le même ID de session seront traités par ordre d'arrivée par un seul consommateur à la fois. .

2. Utilisez le SDK Azure pour gérer les sessions dans Spring Boot

Azure fournit des bibliothèques pour Java qui permettent de consommer les messages avec les sessions de manière ordonnée. Voici une approche :

Ajouter une dépendance Azure Service Bus

Ajoutez la dépendance pour Azure Service Bus dans votre projet Spring Boot (dans pom.xml pour Maven) :

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-servicebus</artifactId>
    <version>7.5.0</version> <!-- check for the last version -->
</dependency>

Configurer le Service Bus dans l'application

Configurez l'application pour se connecter au sujet Azure Service Bus. Voici une configuration de base :

import com.azure.messaging.servicebus.*;

@Service
public class AzureServiceBusConsumer {

    private final String connectionString = "Endpoint=sb://<your-service-bus>.servicebus.windows.net/;SharedAccessKeyName=<key-name>;SharedAccessKey=<key>";
    private final String topicName = "<your-topic>";
    private final String subscriptionName = "<your-subscription>";

    public void startSessionProcessor() {
        ServiceBusClientBuilder clientBuilder = new ServiceBusClientBuilder()
                .connectionString(connectionString);

        ServiceBusProcessorClient processorClient = clientBuilder
                .sessionProcessor() // Using session mode
                .topicName(topicName)
                .subscriptionName(subscriptionName)
                .processMessage(this::processMessage)
                .processError(this::processError)
                .buildProcessorClient();

        // Start session processing in asynchronous mode
        processorClient.start();
    }

    private void processMessage(ServiceBusReceivedMessageContext context) {
        ServiceBusReceivedMessage message = context.getMessage();
        System.out.printf("Processing message from session: %s. Contents: %s%n",
                message.getSessionId(), message.getBody());

        // Process the message here, respecting the order of arrival in the session
        context.complete(); // Mark the message as processed
    }

    private void processError(ServiceBusErrorContext context) {
        System.err.printf("Error occurred while processing: %s%n", context.getException().getMessage());
    }
}

Garantir une consommation unique par session

L'utilisation de sessionProcessor() garantit que chaque session est consommée par une seule instance à la fois et que les messages d'une session sont toujours traités dans l'ordre FIFO, quel que soit le nombre d'instances.

3. Gestion de la concurrence entre les instances EC2

Lorsque plusieurs instances EC2 de l'application sont connectées au sujet :

• Azure Service Bus attribue dynamiquement des sessions aux consommateurs disponibles.
• Chaque instance Spring Boot reçoit des messages une session à la fois, et Azure Service Bus garantit qu'une session active ne sera pas attribuée simultanément à plusieurs consommateurs.
• Cela permet une mise à l'échelle automatique, chaque instance prenant en charge différentes sessions en parallèle.

4. Déploiement et résilience

Déployez l'application Spring Boot sur vos instances EC2 avec des configurations Service Bus identiques, afin qu'elles puissent gérer dynamiquement les sessions. Si une instance tombe en panne, Azure Service Bus réaffectera automatiquement la session en attente à une autre instance EC2 connectée.

5. Exemple de lancement dans Spring Boot

Lancez le consommateur au démarrage de votre application Spring Boot, en utilisant @PostConstruct pour démarrer la consommation en session :

<dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-servicebus</artifactId>
    <version>7.5.0</version> <!-- check for the last version -->
</dependency>

Conclusion

En conclusion, l'intégration d'Azure Service Bus à une application Spring Boot pour le traitement des messages FIFO peut être réalisée efficacement en utilisant des sessions. En activant les sessions sur le sujet Azure Service Bus et en associant les messages à un ID de session spécifique, vous vous assurez que les messages sont traités dans l'ordre exact dans lequel ils arrivent, au sein de chaque session.

À l'aide du SDK Azure pour Java, l'application Spring Boot peut être configurée pour consommer les messages en fonction de la session, garantissant que chaque session est traitée par un seul consommateur à la fois. Cela élimine le risque de réorganisation des messages, même dans un environnement multithread, garantissant un traitement fiable et ordonné.

Cette approche offre une solution évolutive et résiliente, garantissant que l'application traite les messages dans un ordre FIFO strict tout en conservant l'efficacité et la flexibilité dans la gestion des charges de travail distribuées.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!