Utilisation d'Azure Service Bus à partir de Java

Bonjour chers collègues! Il se trouve que notre application est écrite dans la pile java, mais est hébergée dans Azure. Et nous essayons de tirer le meilleur parti des services de gestion du fournisseur de cloud.



L'un d'eux est Azure Service Bus, et aujourd'hui je souhaite parler des fonctionnalités de son utilisation dans une application Spring Boot standard.



Si vous voulez en savoir plus sur les fonctionnalités du râteau, faites défiler jusqu'à la fin de l'article

Qu'est-ce qu'Azure Service Bus?

Quelques mots sur Azure Service Bus est un courtier de messages cloud (remplacement cloud de RabbitMQ, ActiveMQ). Prend en charge les files d'attente (le message est remis à un destinataire) et les sujets (le mécanisme de publication / abonnement) - plus en détail ici, le



support est déclaré:

1. Messages ordonnés - la documentation dit qu'il s'agit d'un FIFO, MAIS il est implémenté en utilisant le concept de sessions de messages - un groupe de messages, pas la file d'attente entière. Si vous devez garantir l'ordre des messages, vous combinez les messages dans un groupe, et maintenant les messages du groupe seront remis en tant que FIFO. Ainsi, Azure Service Bus Queue n'est pas un FIFO - il délivre vos messages aussi aléatoirement qu'il convient
2. File d'attente de lettres mortes - tout est simple ici, ils n'ont pas réussi à livrer le message après N tentatives ou une période de temps - déplacé vers DLQ
3. Livraison planifiée - vous pouvez définir un délai avant la livraison
4. Report de message - masque les messages dans la file d'attente, le message ne sera pas remis automatiquement, mais il peut être récupéré par ID. Nous devons stocker cet identifiant quelque part

Comment s'intégrer à Azure Service Bus

Azure Service Bus prend en charge AMQP 1.0, ce qui signifie qu'il n'est pas compatible avec les clients RabbitMQ. bunny utilise AMQP 0.9.1



Le seul client "standard" qui peut fonctionner avec le Service Bus est Apache Qpid .



Il existe 3 façons de coupler votre application Spring Boot avec Service Bus:

1. JMS + QPID — , — QPID — .
   
   timeout producer — — factory.setCacheProducers(false);
2. Spring Cloud — Azure Service Bus — , . Service Bus
   
   ( 1.2.6) — , azure service bus java sdk.
   
   
   
   Spring Integration — , «Scheduled delivery» «Message deferral» .
   
   
   
   sdk, MessageAndSessionPump
   
   
3. azure service bus java sdk — ,

Spring Cloud — Azure Service Bus

Je vais m'attarder sur cette méthode plus en détail et vous parler des fonctionnalités de l'utilisation de l'

exemple d'application se trouvant dans le référentiel officiel, il est donc inutile de dupliquer le code - le référentiel avec un exemple est ici .



Parce que c'est Spring Integration Messaging, tout se résume à Channel, MessageHandler, MessagingGateway, ServiceActivator.



Et puis il y a le ServiceBusQueueTemplate .

Envoi de messages

Nous devons avoir un canal dans lequel nous écrivons le message que nous voulons envoyer, à l'autre extrémité il y a un MessageHandler qui l'envoie au bus de service.



Le MessagHandler est com.microsoft.azure.spring.integration.core.DefaultMessageHandler - il s'agit du connecteur vers le service externe.



Comment le lier à un canal? - ajoutez l'annotation - @ServiceActivator (inputChannel = OUTPUT_CHANNEL) et maintenant notre MessagHandler écoute le canal OUTPUT_CHANNEL .



Ensuite, nous devons en quelque sorte écrire notre message sur le canal - ici encore la magie du printemps - nous annonçons MessagingGateway et le lions au canal par son nom.



Un extrait de l' exemple :

@MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
public interface QueueOutboundGateway {
    void send(String text);
}

C'est tout: Gateway -> Channel -> MessagHandler -> ServiceBusQueueTemplate -> ServiceBusMessageConverter .



Dans le code, il reste à injecter notre passerelle et à appeler la méthode send .



J'ai mentionné ServiceBusMessageConverter dans la chaîne d'appels pour une raison - si vous souhaitez ajouter des en-têtes personnalisés (par exemple CORRELATION_ID) au message, c'est l'endroit où ils doivent être déplacés de org.springframework.messaging.MessageHeaders vers le message azure.

La méthode spéciale setCustomHeaders .



Dans ce cas, votre passerelle ressemblera à ceci:

@MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
public interface QueueOutboundGateway {
    void send(@Payload String text, @Header("CORRELATION_ID") String correlationId);
}

Recevoir des messages

D'accord, nous savons comment envoyer des messages, comment les obtenir maintenant?



Ici, tout est pareil - MessageProducer -> Channel -> Handler



Le MessageProducer est com.microsoft.azure.spring.integration.servicebus.inbound.ServiceBusQueueInboundChannelAdapter - c'est notre connecteur vers un service externe. À l'intérieur, il y a le même ServiceBusQueueTemplate avec ServiceBusMessageConverter où vous pouvez lire les en-têtes personnalisés et les placer dans le message d'intégration de printemps.



Le canal y est déjà installé à la main:

@Bean
public ServiceBusQueueInboundChannelAdapter queueMessageChannelAdapter(
        @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel, 
        ServiceBusQueueOperation queueOperation) {
    queueOperation.setCheckpointConfig(CheckpointConfig.builder().checkpointMode(CheckpointMode.MANUAL).build());
    ServiceBusQueueInboundChannelAdapter adapter = new ServiceBusQueueInboundChannelAdapter(QUEUE_NAME,
            queueOperation);
    adapter.setOutputChannel(inputChannel);
    return adapter;
}

Mais le Handler lui-même est attaché au canal via @ServiceActivator .

@ServiceActivator(inputChannel = INPUT_CHANNEL)
public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
    String message = new String(payload);
.......

Vous pouvez immédiatement obtenir la ligne:

@ServiceActivator(inputChannel = INPUT_CHANNEL)
public void messageReceiver(String payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
.......

Vous avez peut-être remarqué l'étrange paramètre Checkpointer checkpointer , il est utilisé pour accuser réception manuellement du traitement des messages.

Si vous définissez CheckpointMode.MANUAL lors de la création du ServiceBusQueueInboundChannelAdapter , vous devez envoyer vous-même un accusé de réception du message. Si vous utilisez CheckpointMode.RECORD, la confirmation sera envoyée automatiquement - détails dans le code ServiceBusQueueTemplate .

Caractéristiques d'utilisation

Donc, la liste des "rakes" et "chips" sur lesquels nous sommes déjà allés.

ReceiveMode.PEEKLOCK

Azure Service Bus prend en charge le mode PEEKLOCK - le consommateur prend un message, il se verrouille dans le bus de service, est inaccessible à quiconque pendant un certain temps (durée de verrouillage), mais n'en est pas supprimé. Si dans le délai imparti le consommateur n'a pas envoyé de confirmation de traitement - succès / abandon ou n'a pas prolongé le verrouillage - le message est considéré à nouveau disponible et une nouvelle tentative de livraison sera effectuée.



Fait intéressant, l' abandon réinitialise simplement le verrou et le message devient instantanément disponible pour une redistribution.



ServiceBusQueueTemplate défaut crée QueueClient le mode ReceiveMode.PEEKLOCK .



Si une exception non gérée vole dans notre gestionnaire- aucun accusé de réception ne sera envoyé au serveur et le message restera verrouillé et sera renvoyé avant l'expiration du délai.

Dans ce cas, le compteur de livraison augmentera, ce qui est logique.



Je ne sais pas s'il s'agit d'un bogue ou d'une fonctionnalité - mais il est très pratique de faire un délai entre les tentatives pour les situations où cela est nécessaire.



Si le message ne peut pas être traité même avec une nouvelle tentative, il est nécessaire d'intercepter les exceptions et de marquer le message comme traité et d'ajouter une logique supplémentaire à l'application, sinon il sera remis encore et encore jusqu'à ce qu'il atteigne la limite du nombre de re-remise (configuré lors de la création d'une file d'attente dans le bus de service )

Nombre de messages d'accès concurrentiel et de prélecture

Comme vous l'avez peut-être deviné, le paramètre d' accès concurrentiel est responsable du nombre de gestionnaires de messages parallèles et le nombre de messages de prélecture correspond au nombre de messages que nous allons entrer dans la mémoire tampon à partir du serveur.



Par défaut, le ServiceBusQueueTemplate est configuré automatiquement (AzureServiceBusQueueAutoConfiguration) avec une valeur de 1 pour les deux paramètres, c'est-à-dire par défaut, chaque file d'attente aura un thread de traitement, bien que le concept d'un bus de service avec accusé de réception pour chaque message individuel implique de nombreux processeurs simultanés. Ceci est d'autant plus important si vous avez un long traitement de demande.



Malheureusement, ces paramètres ne peuvent pas être définis via la configuration de l'application (application.yml / application.properties) et ne peuvent être définis que dans le code. Mais même via le code, vous ne pourrez pas définir différents paramètres pour différentes files d'attente.



Par conséquent, si vous devez définir des paramètres différents, vous devrez créer plusieurs beans ServiceBusQueueTemplate pour chaque ServiceBusQueueInboundChannelAdapter

CompletableFuture à l'intérieur d'Azure Service Bus Java SDK

Le sdk java de bus de service azure lui-même est implémenté autour de l' exécuteur CompletableFuture et CachedThreadPool - MessagingFactory.INTERNAL_THREAD_POOL, alors soyez prudent avec toutes sortes de threads locaux.

Messages ordonnés

Nous utilisons le bus de service comme file d'attente de travaux - certains travaux dépendent les uns des autres et doivent donc être exécutés dans l'ordre de leur création.



Comme je l'ai mentionné ci-dessus, les T-shirts utilisent le concept de sessions de messages - lorsque les messages sont regroupés dans une session par clé (transmis dans l'en-tête), la session existe tant qu'il y a au moins un message avec la clé de session - en détail dans la documentation

Service bus garantit la livraison des messages au sein d'un tel groupe dans l'ordre d'ajout à serveur (c'est-à-dire dans l'ordre dans lequel le serveur de bus de service les a écrits dans le référentiel).



Il convient également de mentionner si vous avez créé une file d'attente pour les sessions activées - cela signifie que tous les messages doivent avoir un en-tête avec une clé de session.



Immédiatement, nous avons été très satisfaits de la possibilité du bus de service d'aligner les messages dans une file d'attente FIFO - mais pour un groupe de messages.



Mais après un certain temps, nous avons commencé à remarquer des problèmes:

• certains messages ont commencé à arriver un nombre infini de fois
• le traitement de la file d'attente a ralenti
• dans les statistiques du bus de service, la moitié des demandes sont marquées comme ayant échoué et les demandes ayant échoué apparaissent même dans une file d'attente vide lorsqu'elles sont inactives

En regardant dans le code sdk, nous avons découvert la particularité de travailler avec des sessions:

1. le consommateur capture la session et commence à lire tous les messages disponibles
2. traité simultanément le nombre de sessions égal au paramètre de concurrence
3. unhandled exception — 1 ( ) — re-delivery ? 0 exception — ttl .
4. — success abandon. — delay re-delivery
   
   .. abandon — , delivery counter .
   
   delivery count

En conséquence, ils ont abandonné cette fonction de bus de service et ont écrit un vélo, et le bus de service agit comme un déclencheur.



Dès que la file d'attente des sessions activées a été annulée, les erreurs dans les statistiques ont disparu; la requête au bus de service.



Dans le bundle JMS + Qpid - cette fonctionnalité n'est pas disponible.

Problèmes potentiels avec des tailles de file d'attente supérieures à 1G

Je n'ai pas encore rencontré, mais j'ai entendu dire qu'il commence à fonctionner instable si la taille de la file d'attente est supérieure à 1G.



Si vous rencontrez ceci ou vice versa, tout fonctionne - écrivez dans les commentaires.

Problèmes avec les demandes de traçage

L'agent Azure Application Insights standard ne sait pas comment suivre l'envoi de messages en tant que dépendance et les messages entrants en tant que demandes.



J'ai dû ajouter du code.

Résultat

Si vous avez besoin d'une file d'attente de travaux avec un temps de traitement des messages long et que vous n'avez pas besoin d'une file d'attente, vous pouvez utiliser.



Si le traitement des messages est rapide - utilisez Azure Event Hub - Kafka standard, le client standard fonctionne correctement.