Prise en charge de Spring Integration par Spring Cloud Azure

S’applique à: ✅ Version 4.20.0, ✅ Version 5.25.0, ✅ Version 6.5.0, ✅ Version 7.4.0

Spring Integration Extension pour Azure fournit des adaptateurs Spring Integration pour les différents services fournis par le sdk Azure pour Java. Nous fournissons la prise en charge de Spring Integration pour ces services Azure : Event Hubs, Service Bus, File d’attente de stockage. 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 de Spring à Azure Event Hubs

Concepts clés

Azure Event Hubs est une plateforme de streaming 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 de n’importe quel fournisseur d’analyse en temps réel ou d’adaptateurs de traitement par lots/adaptateurs de stockage.

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 plus élevé par rapport aux mécanismes d’invocation à distance, de messagerie et de planification des tâches offerts par Spring. Le projet d’extension Spring Integration pour Event Hubs fournit des adaptateurs de canal entrants et sortants et des passerelles pour Azure Event Hubs.

Remarque

Les API de prise en charge de RxJava sont abandonnées à partir de la version 4.0.0. Pour plus d’informations, consultez Javadoc.

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 offsets validés dans le broker, vous devez stocker manuellement les offsets des messages d’Event Hubs en cours de traitement. Le Kit de développement logiciel (SDK) Event Hubs fournit la fonction pour stocker ces décalages dans le stockage Azure.

Prise en charge du partitionnement

Event Hubs fournit un concept similaire de partition physique comme Kafka. Mais contrairement au rééquilibrage automatique de Kafka entre consommateurs et partitions, Event Hubs propose une sorte de mode préemptif. Le compte de stockage sert de bail pour déterminer quelle partition appartient à quel consommateur. Lorsqu’un nouveau consommateur démarre, il tente de voler certaines partitions aux consommateurs les plus chargés afin d’équilibrer la 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 de EventHubsContainerProperties.

Prise en charge de la consommation par lots

Le EventHubsInboundChannelAdapter prend 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 instance de EventHubsInboundChannelAdapter. Lorsqu’elle est activée, un message dont la charge utile est une liste d’événements traités par lot sera reçu 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 communs d’ID de partition, de point de contrôle et de dernières propriétés mises en file d’attente, une seule valeur est présentée pour l’ensemble du lot d’événements qui partagent cette même valeur. Pour plus d’informations, consultez la section Des en-têtes de message Event Hubs.

Remarque

L’en-tête du point de contrôle n’existe que lorsque le mode de point de contrôle MANUEL est utilisé.

Le point de contrôle du consommateur de lots prend en charge deux modes : BATCH et MANUAL. Le mode BATCH est un mode de point de contrôle automatique qui permet d’effectuer un point de contrôle sur l’ensemble du lot d’événements en une seule fois dès leur réception. Le mode MANUAL sert à enregistrer les événements des utilisateurs. Lorsqu’il est utilisé, le Checkpointer sera transmis dans l’en-tête du message, et les utilisateurs pourront s’en servir pour effectuer le checkpointing.

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

Configuration des dépendances

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

Paramétrage

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

Propriétés de configuration de connexion

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

Remarque

Si vous choisissez d’utiliser un principal de sécurité pour authentifier et autoriser avec Microsoft Entra ID pour accéder à une ressource Azure, consultez Autoriser l’accès avec l’ID Microsoft Entra pour vous assurer que le principal de sécurité a reçu l’autorisation suffisante pour accéder à la ressource Azure.

Propriétés configurables de connexion de spring-cloud-azure-starter-integration-eventhubs:

Propriété	Type	Descriptif
`spring.cloud.azure.eventhubs.enabled`	booléen	Indique si azure Event Hubs est activé.
`spring.cloud.azure.eventhubs.connection-string`	Chaîne	Valeur de chaîne de connexion de l’espace de noms Event Hubs.
`spring.cloud.azure.eventhubs.namespace`	Chaîne	Valeur de l’espace de noms Event Hubs, qui est le préfixe du FQDN. Un FQDN doit se composer de NamespaceName.DomainName
`spring.cloud.azure.eventhubs.domain-name`	Chaîne	Nom de domaine d’une valeur d’espace de noms Azure Event Hubs.
`spring.cloud.azure.eventhubs.custom-endpoint-address`	Chaîne	Adresse de point de terminaison personnalisée.
`spring.cloud.azure.eventhubs.shared-connection`	Booléen	Indique si EventProcessorClient et EventHubProducerAsyncClient sous-jacents utilisent la même connexion. Par défaut, une nouvelle connexion est construite et utilisée pour chaque client Event Hub créé.

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

Cette section contient les options de configuration pour le service Blobs de stockage, qui est utilisé pour conserver la propriété de partition et les informations de point de contrôle.

Remarque

À partir de la version 4.0.0, lorsque la propriété de spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists n’est pas activée manuellement, aucun conteneur de stockage n’est créé automatiquement.

Propriétés configurables du point de contrôle de spring-cloud-azure-starter-integration-eventhubs :

Propriété	Type	Descriptif
`spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists`	Booléen	Indique s’il faut autoriser la création de conteneurs s’ils n’existent pas.
`spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name`	Chaîne	Nom du compte de stockage.
`spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key`	Chaîne	Clé d’accès au compte de stockage.
`spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name`	Chaîne	Nom du conteneur de stockage.

Les options de configuration courantes du SDK de service Azure peuvent également être configurées pour le magasin de points de contrôle Blob Storage. Les options de configuration prises en charge sont présentées dans Spring Cloud Azure configuration et peuvent être configurées en utilisant soit le préfixe unifié spring.cloud.azure., soit le préfixe spring.cloud.azure.eventhubs.processor.checkpoint-store..

Propriétés de configuration du processeur Event Hub

L'EventHubsInboundChannelAdapter utilise le EventProcessorClient pour consommer des messages à partir d’un hub d’événements, pour 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 de EventHubsInboundChannelAdapter.

Utilisation de base

Envoyer des messages à Azure Event Hubs

Renseignez les options de configuration des informations d’identification.
- Pour les informations d’identification sous forme de chaîne de connexion, configurez les propriétés suivantes dans votre fichier application.yml :
```
spring:
  cloud:
    azure:
      eventhubs:
        connection-string: ${AZURE_EVENT_HUBS_CONNECTION_STRING}
        processor:
          checkpoint-store:
            container-name: ${CHECKPOINT-CONTAINER}
            account-name: ${CHECKPOINT-STORAGE-ACCOUNT}
            account-key: ${CHECKPOINT-ACCESS-KEY}
```
  Remarque
  
  Microsoft recommande d’utiliser le flux d’authentification le plus sécurisé disponible. Le flux d'authentification décrit dans cette procédure, comme pour les bases de données, les caches, la messagerie ou les services IA, demande un degré de confiance très élevé dans l'application et comporte des risques non présents dans d'autres flux. Utilisez ce flux uniquement lorsque des options plus sécurisées, telles que les identités managées pour les connexions sans mot de passe ou sans clé, ne sont pas viables. Pour les opérations d’ordinateur local, préférez les identités utilisateur pour les connexions sans mot de passe ou sans clé.
- 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_EVENT_HUBS_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_EVENT_HUBS_NAMESPACE}
        processor:
          checkpoint-store:
            container-name: ${CONTAINER_NAME}
            account-name: ${ACCOUNT_NAME}
```

Remarque

Les valeurs autorisées pour tenant-id sont : common, organizations, consumersou l’ID de locataire. Pour plus d’informations sur ces valeurs, consultez la section Utilisation du mauvais point de terminaison (comptes personnels et d’entreprise) de Erreur AADSTS50020 - Le compte utilisateur du fournisseur d’identité n’existe pas dans le locataire. Pour plus d’informations sur la conversion de votre application monolocataire, consultez Convertir une application monolocataire en multilocataire sur Microsoft Entra ID.

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

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

Envoyez des messages à l’aide de la passerelle.

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

Recevoir des messages d’Azure Event Hubs

Renseignez les options de configuration des informations d’identification.

Créez un bean de canal de message comme canal d’entrée.

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

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

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 de la consommation par lots

Pour consommer des messages depuis Event Hubs par lots, le processus est similaire à celui de l’exemple ci-dessus, mais les utilisateurs doivent en outre définir pour EventHubsInboundChannelAdapter les options de configuration liées au traitement par lots.

Lorsque vous créez EventHubsInboundChannelAdapter, le mode écouteur doit être défini comme BATCH. Lors de la création du bean de EventHubsMessageListenerContainer, définissez le mode de point de contrôle sur MANUAL ou BATCH ; les options de traitement par lots peuvent être configurées selon vos 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(listenerContainer, 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 des 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é event.

Correspondance entre les messages / propriétés d’événement d’Event Hubs et les en-têtes de message Spring en mode écouteur d’enregistrement :

Propriétés d’événement Event Hubs	Constantes d’en-tête de message Spring	Type	Descriptif
Heure en file d’attente	EventHubsHeaders#ENQUEUED_TIME	Instantané	Instant, en UTC, de l’heure à laquelle l’événement a été mis en file d’attente dans la partition Event Hub.
Compenser	EventHubsHeaders#OFFSET	Long	L’offset de l’événement au moment où il a été reçu depuis la partition Event Hub associée.
Clé de partition	AzureHeaders#PARTITION_KEY	Chaîne	Clé de hachage de partition si elle a été définie lors de la publication initiale de cet événement.
Identifiant de partition	AzureHeaders#RAW_PARTITION_ID	Chaîne	ID de partition du hub d’événements.
Numéro de séquence	EventHubsHeaders#SEQUENCE_NUMBER	Long	Numéro de séquence affecté à l’événement lorsqu’il a été mis en file d’attente dans la partition Event Hub associée.
Propriétés du dernier événement mis en file d’attente	EventHubsHeaders#LAST_ENQUEUED_EVENT_PROPERTIES	LastEnqueuedEventProperties	Propriétés du dernier événement mis en file d’attente dans cette partition.
NA	AzureHeaders#CHECKPOINTER	Point de contrôle	En-tête permettant d’effectuer un point de contrôle sur le message spécifié.

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.

Remarque

Les en-têtes de message suivants ne peuvent pas être définis manuellement : la clé de partition, l’heure de mise en file d’attente, l’offset et le numéro de séquence.

Lorsque le mode de consommateur par lots est activé, les en-têtes spécifiques des messages traités par lot sont répertoriés comme suit ; ils contiennent une liste de valeurs provenant de chaque événement Event Hubs individuel.

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

Propriétés d’événement Event Hubs	Constantes d’en-tête de message Spring Batch	Type	Descriptif
Heure en file d’attente	EventHubsHeaders#ENQUEUED_TIME	Liste d’Instant	Liste de l’instant, au format UTC, de l’heure à laquelle chaque événement a été mis en file d’attente dans la partition Event Hub.
Compenser	EventHubsHeaders#OFFSET	Liste de Long	Liste du décalage de chaque événement lorsqu’il a été reçu de la partition Event Hub associée.
Clé de partition	AzureHeaders#PARTITION_KEY	Liste de chaînes de caractères	Liste de la clé de hachage de partition si elle a été définie lors de la publication initiale de chaque événement.
Numéro de séquence	EventHubsHeaders#SEQUENCE_NUMBER	Liste de Long	Liste du numéro de séquence affecté à chaque événement lorsqu’il a été mis en file d’attente dans la partition Event Hub associée.
Propriétés système	EventHubsHeaders#BATCH_CONVERTED_SYSTEM_PROPERTIES	Liste des cartes	Liste des propriétés système de chaque événement.
Propriétés de l’application	EventHubsHeaders#BATCH_CONVERTED_APPLICATION_PROPERTIES	Liste de Map	Liste des propriétés d’application de chaque événement, où tous les en-têtes de message personnalisés ou propriétés d’événement sont placés.

Remarque

Lors de la publication des messages, tous les en-têtes de lot ci-dessus seront supprimés des messages s’ils existent.

Échantillons

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

Intégration de Spring avec 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.

Remarque

Les API prenant en charge CompletableFuture sont dépréciées depuis la version 2.10.0 et ont été remplacées par Reactor Core à partir de la version 4.0.0. Pour plus d’informations, consultez Javadoc.

Configuration des dépendances

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

Paramétrage

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

Propriétés de configuration de connexion

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

Remarque

Propriétés configurables de connexion de spring-cloud-azure-starter-integration-servicebus:

Propriété	Type	Descriptif
`spring.cloud.azure.servicebus.enabled`	booléen	Indique si Azure Service Bus est activé.
`spring.cloud.azure.servicebus.connection-string`	Chaîne	Valeur de chaîne de connexion de l’espace de noms Service Bus.
`spring.cloud.azure.servicebus.custom-endpoint-address`	Chaîne	Adresse de point de terminaison personnalisée à utiliser lors de la connexion à Service Bus.
`spring.cloud.azure.servicebus.namespace`	Chaîne	Valeur de l’espace de noms de Service Bus, qui constitue le préfixe du FQDN. Un FQDN doit se composer de NamespaceName.DomainName
`spring.cloud.azure.servicebus.domain-name`	Chaîne	Nom de domaine d’une valeur d’espace de noms Azure Service Bus.

Propriétés de configuration du processeur Service Bus

Le ServiceBusInboundChannelAdapter utilise le ServiceBusProcessorClient pour consommer des messages, pour 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 de ServiceBusInboundChannelAdapter.

Utilisation de base

Envoyer des messages à Azure Service Bus

Renseignez les options de configuration des informations d’identification.
- Pour les informations d’identification sous forme de 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}
```
  Remarque
  
  Microsoft recommande d’utiliser le flux d’authentification le plus sécurisé disponible. Le flux d'authentification décrit dans cette procédure, comme pour les bases de données, les caches, la messagerie ou les services IA, demande un degré de confiance très élevé dans l'application et comporte des risques non présents dans d'autres flux. Utilisez ce flux uniquement lorsque des options plus sécurisées, telles que les identités managées pour les connexions sans mot de passe ou sans clé, ne sont pas viables. Pour les opérations d’ordinateur local, préférez les identités utilisateur pour les connexions sans mot de passe ou sans clé.
- 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}
```

Remarque

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}

Remarque

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.error("There was an error sending the message.", ex);
            }
        });

        return handler;
    }
}

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

Envoyez des messages à l’aide de la passerelle.

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

Recevoir des messages d’Azure Service Bus

Renseignez les options de configuration des informations d’identification.

Créez un bean de canal de message comme canal d’entrée.

@Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";

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

Créez ServiceBusInboundChannelAdapter avec le bean ServiceBusMessageListenerContainer pour recevoir des messages depuis 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;
    }
}

Créez une liaison de récepteur de messages avec ServiceBusInboundChannelAdapter 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()
                .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 des messages Service Bus

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

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

En-têtes et propriétés des messages Service Bus	Constantes d’en-tête de message Spring	Type	Paramétrable	Descriptif
Type de contenu	`MessageHeaders#CONTENT_TYPE`	Chaîne	Oui	Descripteur de type de contenu du message RFC2045.
ID de corrélation	`ServiceBusMessageHeaders#CORRELATION_ID`	Chaîne	Oui	ID de corrélation du message
ID de message	`ServiceBusMessageHeaders#MESSAGE_ID`	Chaîne	Oui	L’ID de message du message, cet en-tête a une priorité supérieure à `MessageHeaders#ID`.
ID de message	`MessageHeaders#ID`	UUID (Identifiant Unique Universel)	Oui	L’ID de message du message, cet en-tête a une priorité inférieure à `ServiceBusMessageHeaders#MESSAGE_ID`.
Clé de partition	`ServiceBusMessageHeaders#PARTITION_KEY`	Chaîne	Oui	Clé de partition pour l’envoi du message à une entité partitionnée.
Répondre à	`MessageHeaders#REPLY_CHANNEL`	Chaîne	Oui	Adresse d’une entité à laquelle envoyer des réponses.
Répondre à l’ID de session	`ServiceBusMessageHeaders#REPLY_TO_SESSION_ID`	Chaîne	Oui	Valeur de propriété ReplyToGroupId du message.
Heure planifiée de mise en file d’attente UTC	`ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME`	OffsetDateTime	Oui	Date/heure à laquelle le message doit être mis en file d’attente dans Service Bus, cet en-tête a une priorité supérieure à `AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE`.
Heure planifiée de mise en file d’attente UTC	`AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE`	Entier	Oui	Date/heure à laquelle le message doit être mis en file d’attente dans Service Bus, cet en-tête a une priorité inférieure à `ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME`.
ID de la session	`ServiceBusMessageHeaders#SESSION_ID`	Chaîne	Oui	L’identifiant de session d’une entité prenant en charge les sessions.
Durée de vie	`ServiceBusMessageHeaders#TIME_TO_LIVE`	Durée	Oui	Durée avant l’expiration de ce message.
À	`ServiceBusMessageHeaders#TO`	Chaîne	Oui	Adresse « à » du message, réservée à une utilisation ultérieure dans les scénarios de routage et actuellement ignorée par le répartiteur lui-même.
Objet	`ServiceBusMessageHeaders#SUBJECT`	Chaîne	Oui	Objet du message.
Description de l’erreur de lettre morte	`ServiceBusMessageHeaders#DEAD_LETTER_ERROR_DESCRIPTION`	Chaîne	Non	Description d’un message qui a été placé dans la file d’attente des messages non distribuables.
Raison de la lettre morte	`ServiceBusMessageHeaders#DEAD_LETTER_REASON`	Chaîne	Non	La raison pour laquelle un message a été mis en lettres mortes.
Source de lettres mortes	`ServiceBusMessageHeaders#DEAD_LETTER_SOURCE`	Chaîne	Non	Entité dans laquelle le message a été mis en lettres mortes.
Nombre de livraisons	`ServiceBusMessageHeaders#DELIVERY_COUNT`	long	Non	Nombre de fois où ce message a été remis aux clients.
Numéro de séquence lors de la mise en file d’attente	`ServiceBusMessageHeaders#ENQUEUED_SEQUENCE_NUMBER`	long	Non	Numéro de séquence mis en file d’attente affecté à un message par Service Bus.
Heure en file d’attente	`ServiceBusMessageHeaders#ENQUEUED_TIME`	OffsetDateTime	Non	Date à laquelle ce message a été mis en file d’attente dans Service Bus.
Expire à	`ServiceBusMessageHeaders#EXPIRES_AT`	OffsetDateTime	Non	Date d’expiration de ce message.
Jeton de verrouillage	`ServiceBusMessageHeaders#LOCK_TOKEN`	Chaîne	Non	Jeton de verrouillage pour le message actuel.
Verrouillé jusqu’au	`ServiceBusMessageHeaders#LOCKED_UNTIL`	OffsetDateTime	Non	Date d’expiration du verrou de ce message.
Numéro de séquence	`ServiceBusMessageHeaders#SEQUENCE_NUMBER`	long	Non	Numéro unique affecté à un message par Service Bus.
État	`ServiceBusMessageHeaders#STATE`	ServiceBusMessageState	Non	État du message, qui peut être actif, différé ou planifié.

Prise en charge des clés de partition

Ce starter prend en charge le partitionnement de Service Bus en permettant de définir la clé de partition et 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!");
    }
}

Déconseillé, mais actuellement pris en charge : utiliser AzureHeaders.PARTITION_KEY comme clé d’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!");
    }
}

Remarque

Lorsque les ServiceBusMessageHeaders.PARTITION_KEY et les AzureHeaders.PARTITION_KEY sont définis dans les en-têtes de message, ServiceBusMessageHeaders.PARTITION_KEY est préférable.

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!");
    }
}

Remarque

Lorsque le ServiceBusMessageHeaders.SESSION_ID est défini dans les en-têtes de message et qu’un en-tête de ServiceBusMessageHeaders.PARTITION_KEY différent est également défini, la valeur de l’ID de session sera finalement utilisée pour remplacer la valeur de la clé de partition.

Personnaliser les propriétés du client Service Bus

Les développeurs peuvent utiliser AzureServiceClientBuilderCustomizer pour personnaliser les propriétés du client Service Bus. L’exemple suivant personnalise la propriété sessionIdleTimeout dans ServiceBusClientBuilder:

@Bean
public AzureServiceClientBuilderCustomizer<ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder> customizeBuilder() {
    return builder -> builder.sessionIdleTimeout(Duration.ofSeconds(10));
}

Échantillons

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

Intégration de Spring avec Azure Queue Storage

Concepts clés

Le stockage de files d’attente Azure est un service qui permet de stocker un grand nombre de messages. Vous accédez à des messages depuis n’importe où dans le monde via des appels authentifiés à l’aide de HTTP ou HTTPS. Un message de file d’attente peut comporter jusqu’à 64 Ko de taille. Une file d’attente peut contenir des millions de messages, jusqu’à la limite totale de capacité d’un compte de stockage. Les files d’attente sont couramment utilisées pour créer un backlog de travail à traiter de manière asynchrone.

Configuration des dépendances

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

Paramétrage

Ce démarrage fournit les options de configuration suivantes :

Propriétés de configuration de connexion

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

Remarque

Propriétés configurables de connexion de spring-cloud-azure-starter-integration-storage-queue:

Propriété	Type	Descriptif
`spring.cloud.azure.storage.queue.enabled`	booléen	Indique si une file d’attente stockage Azure est activée.
`spring.cloud.azure.storage.queue.connection-string`	Chaîne	Valeur de la chaîne de connexion de l’espace de noms de la file d’attente de stockage.
`spring.cloud.azure.storage.queue.accountName`	Chaîne	Nom du compte de file d’attente de stockage.
`spring.cloud.azure.storage.queue.accountKey`	Chaîne	Clé du compte de file d’attente de stockage.
`spring.cloud.azure.storage.queue.endpoint`	Chaîne	Point de terminaison du service de file d’attente de stockage.
`spring.cloud.azure.storage.queue.sasToken`	Chaîne	Informations d’identification du jeton SAS
`spring.cloud.azure.storage.queue.serviceVersion`	QueueServiceVersion	QueueServiceVersion utilisée lors de l’envoi de requêtes d’API.
`spring.cloud.azure.storage.queue.messageEncoding`	Chaîne	Encodage des messages de file d’attente.

Utilisation de base

Envoyer des messages à la file d’attente du stockage Azure

Renseignez les options de configuration des informations d’identification.
- Pour les informations d’identification sous forme de chaîne de connexion, configurez les propriétés suivantes dans votre fichier application.yml :
```
spring:
  cloud:
    azure:
      storage:
        queue:
          connection-string: ${AZURE_STORAGE_QUEUE_CONNECTION_STRING}
```
  Remarque
  
  Microsoft recommande d’utiliser le flux d’authentification le plus sécurisé disponible. Le flux d'authentification décrit dans cette procédure, comme pour les bases de données, les caches, la messagerie ou les services IA, demande un degré de confiance très élevé dans l'application et comporte des risques non présents dans d'autres flux. Utilisez ce flux uniquement lorsque des options plus sécurisées, telles que les identités managées pour les connexions sans mot de passe ou sans clé, ne sont pas viables. Pour les opérations d’ordinateur local, préférez les identités utilisateur pour les connexions sans mot de passe ou sans clé.
- 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:
          account-name: ${AZURE_STORAGE_QUEUE_ACCOUNT_NAME}
```

Remarque

Les valeurs autorisées pour tenant-id sont : common, organizations, consumersou l’ID de locataire. Pour plus d’informations sur ces valeurs, consultez la section Utilisation du mauvais point de terminaison (comptes personnels et organisationnels) de Erreur AADSTS50020 - Le compte d’utilisateur du fournisseur d’identité n’existe pas dans le locataire. Pour plus d’informations sur la conversion de votre application monolocataire, consultez Convertir une application monolocataire en multilocataire sur Microsoft Entra ID.

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:
          account-name: ${AZURE_STORAGE_QUEUE_ACCOUNT_NAME}

Remarque

Créez DefaultMessageHandler avec le bean StorageQueueTemplate pour envoyer des messages vers une file d’attente de stockage.

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.error("There was an error sending the message.", ex);
            }
        });
        return handler;
    }
}

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

Envoyez des messages à l’aide de la passerelle.

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

Recevoir des messages de la file d’attente du stockage Azure

Renseignez les options de configuration des informations d’identification.

Créez un bean de canal de messages comme canal d’entrée.

class Demo {
    private static final String INPUT_CHANNEL = "input";

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

Créez StorageQueueMessageSource avec le bean StorageQueueTemplate pour recevoir des messages depuis une file d’attente de stockage.

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

Créez une liaison de récepteur de messages avec StorageQueueMessageSource 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();
    }
}

Échantillons

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

Commentaires

Cette page a-t-elle été utile ?

Last updated on 2026-09-21

Prise en charge de Spring Integration par Spring Cloud Azure

Intégration de Spring à Azure Event Hubs

Concepts clés

Groupe de consommateurs

Prise en charge du partitionnement

Prise en charge de la consommation par lots

Configuration des dépendances

Paramétrage

Propriétés de configuration de connexion

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

Propriétés de configuration du processeur Event Hub

Utilisation de base

Envoyer des messages à Azure Event Hubs

Recevoir des messages d’Azure Event Hubs

Configurer EventHubsMessageConverter pour personnaliser objectMapper

Prise en charge de la consommation par lots

En-têtes de message des Event Hubs

Échantillons

Intégration de Spring avec Azure Service Bus

Concepts clés

Configuration des dépendances

Paramétrage

Propriétés de configuration de connexion

Propriétés de configuration du processeur Service Bus

Utilisation de base

Envoyer des messages à Azure Service Bus

Recevoir des messages d’Azure Service Bus

Configurer ServiceBusMessageConverter pour personnaliser objectMapper

En-têtes des messages Service Bus

Prise en charge des clés de partition

Prise en charge des sessions

Personnaliser les propriétés du client Service Bus

Échantillons

Intégration de Spring avec Azure Queue Storage

Concepts clés

Configuration des dépendances

Paramétrage

Propriétés de configuration de connexion

Utilisation de base

Envoyer des messages à la file d’attente du stockage Azure

Recevoir des messages de la file d’attente du stockage Azure

Échantillons

Commentaires

Ressources supplémentaires