Spring Cloud support Azure pour Spring Integration

Cet article s’applique à : ✔️ Version 4.14.0 ✔️ Version 5.8.0

Spring Integration Extension pour Azure fournit des adaptateurs Spring Integration pour les différents services fournis par le Kit de développement logiciel (SDK) Azure pour Java. Nous fournissons la prise en charge de Spring Integration pour ces services Azure : Event Hubs, Service Bus, Stockage File d’attente. Voici une liste des adaptateurs pris en charge :

• spring-cloud-azure-starter-integration-eventhubs - Pour plus d’informations, consultez Spring Integration avec Azure Event Hubs
• spring-cloud-azure-starter-integration-servicebus - Pour plus d’informations, consultez Spring Integration avec Azure Service Bus
• spring-cloud-azure-starter-integration-storage-queue- Pour plus d’informations, consultez Spring Integration with Stockage Azure Queue

Intégration spring à Azure Event Hubs

Concepts clés

Azure Event Hubs est une plateforme de streaming de Big Data et un service d’ingestion d’événements. Il peut recevoir et traiter des millions d’événements par seconde. Les données envoyées à un concentrateur d’événements peuvent être transformées et stockées à l’aide d’adaptateurs de traitement par lot/stockage ou d’un fournisseur d’analyse en temps réel.

Spring Integration permet une messagerie légère dans les applications Spring et prend en charge l’intégration avec des systèmes externes via des adaptateurs déclaratifs. Ces adaptateurs fournissent un niveau d’abstraction supérieur sur la prise en charge de Spring pour la communication à distance, la messagerie et la planification. Le projet d’extension Spring Integration pour Event Hubs fournit des adaptateurs de canal entrants et sortants et des passerelles pour Azure Event Hubs.

Groupe de consommateurs

Event Hubs fournit une prise en charge similaire du groupe de consommateurs comme Apache Kafka, mais avec une légère logique différente. Bien que Kafka stocke tous les décalages validés dans le répartiteur, vous devez stocker les décalages des messages Event Hubs traités manuellement. Le Kit de développement logiciel (SDK) Event Hubs fournit la fonction pour stocker ces décalages à l’intérieur de Stockage Azure.

Prise en charge du partitionnement

Event Hubs fournit un concept similaire de partition physique comme Kafka. Mais contrairement à la rééquilibrage automatique de Kafka entre les consommateurs et les partitions, Event Hubs fournit un type de mode préemptif. Le compte de stockage agit en tant que bail pour déterminer la partition détenue par le consommateur. Lorsqu’un nouveau consommateur démarre, il tente de voler certaines partitions de la plupart des consommateurs chargés pour atteindre l’équilibrage de charge de travail.

Pour spécifier la stratégie d’équilibrage de charge, les développeurs peuvent utiliser EventHubsContainerProperties pour la configuration. Consultez la section suivante pour obtenir un exemple de configuration EventHubsContainerProperties.

Prise en charge des consommateurs Batch

Prend EventHubsInboundChannelAdapter en charge le mode de consommation par lots. Pour l’activer, les utilisateurs peuvent spécifier le mode d’écouteur comme ListenerMode.BATCH lors de la construction d’une EventHubsInboundChannelAdapter instance. Lorsqu’elle est activée, un message dont la charge utile est une liste d’événements par lots seront reçus et transmis au canal en aval. Chaque en-tête de message est également converti en tant que liste, dont le contenu est la valeur d’en-tête associée analysée à partir de chaque événement. Pour les en-têtes communaux de l’ID de partition, case activée pointer et les dernières propriétés mises en file d’attente, ils sont présentés comme une valeur unique pour l’ensemble du lot d’événements partagent le même. Pour plus d’informations, consultez la section En-têtes de message Event Hubs .

Le point de contrôle du consommateur de lots prend en charge deux modes : BATCH et MANUAL. BATCHil s’agit d’un mode de case activée point automatique pour case activée pointer l’ensemble du lot d’événements une fois qu’ils sont reçus. MANUALle mode consiste à case activée pointer les événements par les utilisateurs. Lorsqu’il est utilisé, le point de contrôle est passé dans l’en-tête de message, et les utilisateurs peuvent l’utiliser pour effectuer case activée pointage.

La stratégie de consommation de lots peut être spécifiée par les propriétés et max-sizemax-wait-time, où max-size est une propriété nécessaire pendant qu’elle max-wait-time est facultative. Pour spécifier la stratégie de consommation de lots, les développeurs peuvent utiliser EventHubsContainerProperties pour la configuration. Consultez la section suivante pour obtenir un exemple de configuration EventHubsContainerProperties.

Configuration des dépendances

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-integration-eventhubs</artifactId>
</dependency>

Configuration

Ce démarrage fournit les 3 parties suivantes des options de configuration :

Propriétés de configuration de Connecter ion

Cette section contient les options de configuration utilisées pour la connexion à Azure Event Hubs.

propriétés configurables Connecter ion de spring-cloud-azure-starter-integration-eventhubs :

Propriétés de configuration de point de contrôle

Cette section contient les options de configuration du service Stockage Blobs, qui est utilisé pour conserver la propriété de partition et les informations case activée point.

Contrôle des propriétés configurables de spring-cloud-azure-starter-integration-eventhubs :

Les options de configuration courantes du Kit de développement logiciel (SDK) Azure Service sont également configurables pour Stockage magasin d’objets blob case activée point. Les options de configuration prises en charge sont introduites dans la configuration d’Azure Spring Cloud et peuvent être configurées avec le préfixe spring.cloud.azure. unifié ou le préfixe de spring.cloud.azure.eventhubs.processor.checkpoint-store.

Propriétés de configuration du processeur Event Hub

L’utilisation EventHubsInboundChannelAdapter des EventProcessorClient messages à partir d’un hub d’événements permet de configurer les propriétés globales d’un EventProcessorClient, les développeurs peuvent utiliser EventHubsContainerProperties pour la configuration. Consultez la section suivante sur l’utilisation EventHubsInboundChannelAdapterde .

Utilisation de base

Envoyer des messages à Azure Event Hubs

1. Renseignez les options de configuration des informations d’identification.

   • Pour les informations d’identification en tant que chaîne de connexion, configurez les propriétés suivantes dans votre fichier application.yml :

     spring:
       cloud:
         azure:
           eventhubs:
             connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}
             processor:
               checkpoint-store:
                 container-name: ${CHECKPOINT-CONTAINER}
                 account-name: ${CHECKPOINT-STORAGE-ACCOUNT}
                 account-key: ${CHECKPOINT-ACCESS-KEY}
     

   • Pour les informations d’identification en tant qu’identités managées, configurez les propriétés suivantes dans votre fichier application.yml :

     spring:
       cloud:
         azure:
           credential:
             managed-identity-enabled: true
             client-id: ${AZURE_CLIENT_ID}
           eventhubs:
             namespace: ${AZURE_SERVICE_BUS_NAMESPACE}
             processor:
               checkpoint-store:
                 container-name: ${CONTAINER_NAME}
                 account-name: ${ACCOUNT_NAME}
     

   • Pour les informations d’identification en tant que principal de service, configurez les propriétés suivantes dans votre fichier application.yml :

     spring:
       cloud:
         azure:
           credential:
             client-id: ${AZURE_CLIENT_ID}
             client-secret: ${AZURE_CLIENT_SECRET}
           profile:
             tenant-id: <tenant>
           eventhubs:
             namespace: ${AZURE_SERVICE_BUS_NAMESPACE}
             processor:
               checkpoint-store:
                 container-name: ${CONTAINER_NAME}
                 account-name: ${ACCOUNT_NAME}
     

1. Créez DefaultMessageHandler avec le EventHubsTemplate bean pour envoyer des messages à Event Hubs.

   class Demo {
       private static final String OUTPUT_CHANNEL = "output";
       private static final String EVENTHUB_NAME = "eh1";
   
       @Bean
       @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
       public MessageHandler messageSender(EventHubsTemplate eventHubsTemplate) {
           DefaultMessageHandler handler = new DefaultMessageHandler(EVENTHUB_NAME, eventHubsTemplate);
           handler.setSendCallback(new ListenableFutureCallback<Void>() {
               @Override
               public void onSuccess(Void result) {
                   LOGGER.info("Message was sent successfully.");
               }
               @Override
               public void onFailure(Throwable ex) {
                   LOGGER.error("There was an error sending the message.", ex);
               }
           });
           return handler;
       }
   }
   

2. Créez une liaison de passerelle de message avec le gestionnaire de messages ci-dessus via un canal de message.

   class Demo {
       @Autowired
       EventHubOutboundGateway messagingGateway;
   
       @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
       public interface EventHubOutboundGateway {
           void send(String text);
       }
   }
   

3. Envoyez des messages à l’aide de la passerelle.

   class Demo {
       public void demo() {
           this.messagingGateway.send(message);
       }
   }
   

Recevoir des messages d’Azure Event Hubs

1. Renseignez les options de configuration des informations d’identification.

2. Créez unan de canal de message en tant que canal d’entrée.

   @Configuration
   class Demo {
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   }
   

3. Créez EventHubsInboundChannelAdapter avec le EventHubsMessageListenerContainer bean pour recevoir des messages d’Event Hubs.

   @Configuration
   class Demo {
       private static final String INPUT_CHANNEL = "input";
       private static final String EVENTHUB_NAME = "eh1";
       private static final String CONSUMER_GROUP = "$Default";
   
       @Bean
       public EventHubsInboundChannelAdapter messageChannelAdapter(
               @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
               EventHubsMessageListenerContainer listenerContainer) {
           EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(processorContainer);
           adapter.setOutputChannel(inputChannel);
           return adapter;
       }
   
       @Bean
       public EventHubsMessageListenerContainer messageListenerContainer(EventHubsProcessorFactory processorFactory) {
           EventHubsContainerProperties containerProperties = new EventHubsContainerProperties();
           containerProperties.setEventHubName(EVENTHUB_NAME);
           containerProperties.setConsumerGroup(CONSUMER_GROUP);
           containerProperties.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));
           return new EventHubsMessageListenerContainer(processorFactory, containerProperties);
       }
   }
   

4. Créez une liaison de récepteur de messages avec EventHubsInboundChannelAdapter via le canal de message créé précédemment.

   class Demo {
       @ServiceActivator(inputChannel = INPUT_CHANNEL)
       public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
           String message = new String(payload);
           LOGGER.info("New message received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message))
                   .doOnError(e -> LOGGER.error("Error found", e))
                   .block();
       }
   }
   

Configurer EventHubsMessageConverter pour personnaliser objectMapper

EventHubsMessageConverter est fait en tant que bean configurable pour permettre aux utilisateurs de personnaliser ObjectMapper.

Prise en charge des consommateurs Batch

Pour consommer des messages à partir d’Event Hubs par lots, il est similaire à l’exemple ci-dessus, en plus des utilisateurs doivent définir les options de configuration associées au traitement par lots pour EventHubsInboundChannelAdapter.

Lorsque vous créez EventHubsInboundChannelAdapter, le mode écouteur doit être défini sur BATCH. Lors de la création d’unan, EventHubsMessageListenerContainerdéfinissez le mode case activée point en tant que MANUAL ou BATCH, et les options de traitement par lots peuvent être configurées en fonction des besoins.

@Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";
    private static final String EVENTHUB_NAME = "eh1";
    private static final String CONSUMER_GROUP = "$Default";

    @Bean
    public EventHubsInboundChannelAdapter messageChannelAdapter(
            @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
            EventHubsMessageListenerContainer listenerContainer) {
        EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(processorContainer, ListenerMode.BATCH);
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }

    @Bean
    public EventHubsMessageListenerContainer messageListenerContainer(EventHubsProcessorFactory processorFactory) {
        EventHubsContainerProperties containerProperties = new EventHubsContainerProperties();
        containerProperties.setEventHubName(EVENTHUB_NAME);
        containerProperties.setConsumerGroup(CONSUMER_GROUP);
        containerProperties.getBatch().setMaxSize(100);
        containerProperties.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));
        return new EventHubsMessageListenerContainer(processorFactory, containerProperties);
    }
}

En-têtes de message Event Hubs

Le tableau suivant illustre la façon dont les propriétés de message Event Hubs sont mappées aux en-têtes de message Spring. Pour Azure Event Hubs, le message est appelé en tant que event.

Mappage entre event Hubs Message / Propriétés d’événement et en-têtes de message Spring en mode d’écoute d’enregistrement :

Les utilisateurs peuvent analyser les en-têtes de message pour les informations associées de chaque événement. Pour définir un en-tête de message pour l’événement, tous les en-têtes personnalisés sont placés en tant que propriété d’application d’un événement, où l’en-tête est défini comme clé de propriété. Lorsque les événements sont reçus d’Event Hubs, toutes les propriétés de l’application sont converties en en-tête de message.

Lorsque le mode consommateur par lots est activé, les en-têtes spécifiques des messages par lots sont répertoriés ci-dessous, qui contient une liste de valeurs de chaque événement Event Hubs unique.

Mappage entre event Hubs Message / Propriétés d’événement et en-têtes de message Spring en mode Écouteur Batch :

Exemples

Pour plus d’informations, consultez le référentiel azure-spring-boot-samples sur GitHub.

Intégration de Spring à Azure Service Bus

Concepts clés

Spring Integration permet une messagerie légère dans les applications Spring et prend en charge l’intégration avec des systèmes externes via des adaptateurs déclaratifs.

Le projet d’extension Spring Integration pour Azure Service Bus fournit des adaptateurs de canal entrants et sortants pour Azure Service Bus.

Configuration des dépendances

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-integration-servicebus</artifactId>
</dependency>

Configuration

Ce démarrage fournit les 2 parties suivantes des options de configuration :

propriétés de configuration de Connecter ion

Cette section contient les options de configuration utilisées pour la connexion à Azure Service Bus.

propriétés configurables Connecter ion de spring-cloud-azure-starter-integration-servicebus :

Propriétés de configuration du processeur Service Bus

L’utilisation ServiceBusInboundChannelAdapter des ServiceBusProcessorClient messages permet de configurer les propriétés globales d’un ServiceBusProcessorClient, les développeurs peuvent utiliser ServiceBusContainerProperties pour la configuration. Consultez la section suivante sur l’utilisation ServiceBusInboundChannelAdapterde .

Utilisation de base

Envoyer des messages à Azure Service Bus

1. Renseignez les options de configuration des informations d’identification.

   • Pour les informations d’identification en tant que chaîne de connexion, configurez les propriétés suivantes dans votre fichier application.yml :

     spring:
       cloud:
         azure:
           servicebus:
             connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}
     

   • Pour les informations d’identification en tant qu’identités managées, configurez les propriétés suivantes dans votre fichier application.yml :

     spring:
       cloud:
         azure:
           credential:
             managed-identity-enabled: true
             client-id: ${AZURE_CLIENT_ID}
           profile:
             tenant-id: <tenant>
           servicebus:
             namespace: ${AZURE_SERVICE_BUS_NAMESPACE}
     

• Pour les informations d’identification en tant que principal de service, configurez les propriétés suivantes dans votre fichier application.yml :

  spring:
    cloud:
      azure:
        credential:
          client-id: ${AZURE_CLIENT_ID}
          client-secret: ${AZURE_CLIENT_SECRET}
        profile:
          tenant-id: <tenant>
        servicebus:
          namespace: ${AZURE_SERVICE_BUS_NAMESPACE}
  

1. Créez DefaultMessageHandler avec le ServiceBusTemplate bean pour envoyer des messages à Service Bus, définissez le type d’entité pour ServiceBusTemplate. Cet exemple utilise la file d’attente Service Bus comme exemple.

   class Demo {
       private static final String OUTPUT_CHANNEL = "queue.output";
   
       @Bean
       @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
       public MessageHandler queueMessageSender(ServiceBusTemplate serviceBusTemplate) {
           serviceBusTemplate.setDefaultEntityType(ServiceBusEntityType.QUEUE);
           DefaultMessageHandler handler = new DefaultMessageHandler(QUEUE_NAME, serviceBusTemplate);
           handler.setSendCallback(new ListenableFutureCallback<Void>() {
               @Override
               public void onSuccess(Void result) {
                   LOGGER.info("Message was sent successfully.");
               }
   
               @Override
               public void onFailure(Throwable ex) {
                   LOGGER.info("There was an error sending the message.");
               }
           });
   
           return handler;
       }
   }
   

2. Créez une liaison de passerelle de message avec le gestionnaire de messages ci-dessus via un canal de message.

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

3. Envoyez des messages à l’aide de la passerelle.

   class Demo {
       public void demo() {
           this.messagingGateway.send(message);
       }
   }
   

Recevoir des messages d’Azure Service Bus

1. Renseignez les options de configuration des informations d’identification.

2. Créez unan de canal de message en tant que canal d’entrée.

   @Configuration
   class Demo {
       private static final String INPUT_CHANNEL = "input";
   
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   }
   

3. Créez ServiceBusInboundChannelAdapter avec le ServiceBusMessageListenerContainer bean pour recevoir des messages vers Service Bus. Cet exemple utilise la file d’attente Service Bus comme exemple.

   @Configuration
   class Demo {
       private static final String QUEUE_NAME = "queue1";
   
       @Bean
       public ServiceBusMessageListenerContainer messageListenerContainer(ServiceBusProcessorFactory processorFactory) {
           ServiceBusContainerProperties containerProperties = new ServiceBusContainerProperties();
           containerProperties.setEntityName(QUEUE_NAME);
           containerProperties.setAutoComplete(false);
           return new ServiceBusMessageListenerContainer(processorFactory, containerProperties);
       }
   
       @Bean
       public ServiceBusInboundChannelAdapter queueMessageChannelAdapter(
           @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
           ServiceBusMessageListenerContainer listenerContainer) {
           ServiceBusInboundChannelAdapter adapter = new ServiceBusInboundChannelAdapter(listenerContainer);
           adapter.setOutputChannel(inputChannel);
           return adapter;
       }
   }
   

4. Créez une liaison de récepteur de messages via ServiceBusInboundChannelAdapter le canal de message que nous avons créé précédemment.

   class Demo {
       @ServiceActivator(inputChannel = INPUT_CHANNEL)
       public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
           String message = new String(payload);
           LOGGER.info("New message received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message))
                   .doOnError(e -> LOGGER.error("Error found", e))
                   .block();
       }
   }
   

Configurer ServiceBusMessageConverter pour personnaliser objectMapper

ServiceBusMessageConverter est fait en tant que bean configurable pour permettre aux utilisateurs de personnaliser ObjectMapper.

En-têtes de message Service Bus

Pour certains en-têtes Service Bus qui peuvent être mappés à plusieurs constantes d’en-tête Spring, la priorité des différents en-têtes Spring est répertoriée.

Mappage entre les en-têtes Service Bus et les en-têtes Spring :

Prise en charge des clés de partition

Ce démarrage prend en charge le partitionnement Service Bus en autorisant la définition de la clé de partition et de l’ID de session dans l’en-tête du message. Cette section explique comment définir la clé de partition pour les messages.

Recommandé : Utilisez ServiceBusMessageHeaders.PARTITION_KEY comme clé de l’en-tête.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(ServiceBusMessageHeaders.PARTITION_KEY, "Customize partition key")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Non recommandé mais actuellement pris en charge :AzureHeaders.PARTITION_KEY comme clé de l’en-tête.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(AzureHeaders.PARTITION_KEY, "Customize partition key")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Prise en charge des sessions

Cet exemple montre comment définir manuellement l’ID de session d’un message dans l’application.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session ID")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Exemples

Pour plus d’informations, consultez le référentiel azure-spring-boot-samples sur GitHub.

Intégration de Spring au stockage File d’attente Azure

Concepts clés

Stockage File d’attente Azure est un service permettant de stocker un grand nombre de messages. Vous accédez aux messages depuis n’importe où dans le monde par le biais d’appels authentifiés à l’aide du protocole HTTP ou HTTPS. La taille maximale d’un message de file d’attente est de 64 Ko. Une file d’attente peut contenir des millions de messages, dans la limite de la capacité totale d’un compte de stockage. Les files d’attente sont couramment utilisées pour créer un backlog de travail à traiter de façon asynchrone.

Configuration des dépendances

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-integration-storage-queue</artifactId>
</dependency>

Configuration

Ce démarrage fournit les options de configuration suivantes :

propriétés de configuration de Connecter ion

Cette section contient les options de configuration utilisées pour la connexion à Stockage Azure File d’attente.

propriétés configurables Connecter ion de spring-cloud-azure-starter-integration-storage-queue :

Utilisation de base

Envoyer des messages à Stockage Azure File d’attente

1. Renseignez les options de configuration des informations d’identification.

   • Pour les informations d’identification en tant que chaîne de connexion, configurez les propriétés suivantes dans votre fichier application.yml :

     spring:
       cloud:
         azure:
           storage:
             queue:
               connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}
     

   • Pour les informations d’identification en tant qu’identités managées, configurez les propriétés suivantes dans votre fichier application.yml :

     spring:
       cloud:
         azure:
           credential:
             managed-identity-enabled: true
             client-id: ${AZURE_CLIENT_ID}
           profile:
             tenant-id: <tenant>
           storage:
             queue:
               namespace: ${AZURE_SERVICE_BUS_NAMESPACE}
     

• Pour les informations d’identification en tant que principal de service, configurez les propriétés suivantes dans votre fichier application.yml :

  spring:
    cloud:
      azure:
        credential:
          client-id: ${AZURE_CLIENT_ID}
          client-secret: ${AZURE_CLIENT_SECRET}
        profile:
          tenant-id: <tenant>
        storage:
          queue:
            namespace: ${AZURE_SERVICE_BUS_NAMESPACE}
  

1. Créez DefaultMessageHandler avec le StorageQueueTemplate haricot pour envoyer des messages à Stockage File d’attente.

   class Demo {
       private static final String STORAGE_QUEUE_NAME = "example";
       private static final String OUTPUT_CHANNEL = "output";
   
       @Bean
       @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
       public MessageHandler messageSender(StorageQueueTemplate storageQueueTemplate) {
           DefaultMessageHandler handler = new DefaultMessageHandler(STORAGE_QUEUE_NAME, storageQueueTemplate);
           handler.setSendCallback(new ListenableFutureCallback<Void>() {
               @Override
               public void onSuccess(Void result) {
                   LOGGER.info("Message was sent successfully.");
               }
   
               @Override
               public void onFailure(Throwable ex) {
                   LOGGER.info("There was an error sending the message.");
               }
           });
           return handler;
       }
   }
   

2. Créez une liaison de passerelle de message avec le gestionnaire de messages ci-dessus via un canal de message.

   class Demo {
       @Autowired
       StorageQueueOutboundGateway storageQueueOutboundGateway;
   
       @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
       public interface StorageQueueOutboundGateway {
           void send(String text);
       }
   }
   

3. Envoyez des messages à l’aide de la passerelle.

   class Demo {
       public void demo() {
           this.storageQueueOutboundGateway.send(message);
       }
   }
   

Recevoir des messages de Stockage Azure File d’attente

1. Renseignez les options de configuration des informations d’identification.

2. Créez unan de canal de message en tant que canal d’entrée.

   class Demo {
       private static final String INPUT_CHANNEL = "input";
   
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   }
   

3. Créez StorageQueueMessageSource avec le StorageQueueTemplate bean pour recevoir des messages dans Stockage File d’attente.

   class Demo {
       private static final String STORAGE_QUEUE_NAME = "example";
   
       @Bean
       @InboundChannelAdapter(channel = INPUT_CHANNEL, poller = @Poller(fixedDelay = "1000"))
       public StorageQueueMessageSource storageQueueMessageSource(StorageQueueTemplate storageQueueTemplate) {
           return new StorageQueueMessageSource(STORAGE_QUEUE_NAME, storageQueueTemplate);
       }
   }
   

4. Créez une liaison de récepteur de messages avec Stockage QueueMessageSource créée à la dernière étape via le canal de message que nous avons créé précédemment.

   class Demo {
       @ServiceActivator(inputChannel = INPUT_CHANNEL)
       public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
           String message = new String(payload);
           LOGGER.info("New message received: '{}'", message);
           checkpointer.success()
               .doOnError(Throwable::printStackTrace)
               .doOnSuccess(t -> LOGGER.info("Message '{}' successfully checkpointed", message))
               .block();
       }
   }
   

Exemples

Pour plus d’informations, consultez le référentiel azure-spring-boot-samples sur GitHub.