Spring Cloud support Azure pour Spring Cloud Stream

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

Spring Cloud Stream est un framework permettant de créer des microservices hautement évolutifs basés sur des événements connectés à des systèmes de messagerie partagés.

Le framework fournit un modèle de programmation flexible basé sur des idiomes spring déjà établis et familiers et des meilleures pratiques. Ces meilleures pratiques incluent la prise en charge de la sémantique pub/sous-sémantique persistante, des groupes de consommateurs et des partitions avec état.

Les implémentations de classeur actuelles sont les suivantes :

• spring-cloud-azure-stream-binder-eventhubs - Pour plus d’informations, consultez Spring Cloud Stream Binder pour Azure Event Hubs
• spring-cloud-azure-stream-binder-servicebus - pour plus d’informations, consultez Spring Cloud Stream Binder pour Azure Service Bus

Spring Cloud Stream Binder pour Azure Event Hubs

Concepts clés

Spring Cloud Stream Binder pour Azure Event Hubs fournit l’implémentation de liaison pour l’infrastructure Spring Cloud Stream. Cette implémentation utilise les adaptateurs de canal Spring Integration Event Hubs à sa base. Du point de vue de la conception, Event Hubs est similaire à Kafka. En outre, Event Hubs est accessible via l’API Kafka. Si votre projet a une dépendance étroite avec l’API Kafka, vous pouvez essayer Event Hub avec l’exemple d’API Kafka

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 au 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 quel consommateur possède la partition. Lorsqu’un nouveau consommateur démarre, il tente de voler certaines partitions des consommateurs les plus chargés pour atteindre l’équilibre de charge de travail.

Pour spécifier la stratégie d’équilibrage de charge, les propriétés de celles-ci spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.load-balancing.* sont fournies. Pour plus d’informations, consultez la section Propriétés du consommateur.

Prise en charge des consommateurs Batch

Le classeur Spring Cloud Azure Stream Event Hubs prend en charge la fonctionnalité consommateur Spring Cloud Stream Batch.

Pour utiliser le mode batch-consumer, définissez la spring.cloud.stream.bindings.<binding-name>.consumer.batch-mode propriété truesur . Lorsqu’il est activé, un message avec une charge utile d’une liste d’événements par lots est reçu et transmis à la Consumer fonction. Chaque en-tête de message est également converti en liste, dont le contenu est la valeur d’en-tête associée analysée à partir de chaque événement. 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 sont présentés sous forme de valeur unique, car l’ensemble du lot d’événements partage la même valeur. Pour plus d’informations, consultez la section En-têtes de message Event Hubs de Spring Cloud support Azure pour Spring Integration.

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 que le classeur les reçoit. MANUALle mode consiste à case activée pointer les événements par les utilisateurs. Lorsqu’il est utilisé, le Checkpointer message est passé dans l’en-tête de message et les utilisateurs peuvent l’utiliser pour effectuer case activée pointage.

Vous pouvez spécifier la taille du lot en définissant les max-size propriétés qui max-wait-time ont un préfixe spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.batch.. La max-size propriété est nécessaire et la max-wait-time propriété est facultative. Pour plus d’informations, consultez la section Propriétés du consommateur.

Configuration des dépendances

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-stream-binder-eventhubs</artifactId>
</dependency>

Vous pouvez également utiliser Spring Cloud Azure Stream Event Hubs Starter, comme illustré dans l’exemple suivant pour Maven :

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

Configuration

Le classeur fournit les trois 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-stream-binder-eventhubs :

Le classeur prend également en charge Spring Could Azure Resource Manager par défaut. Pour en savoir plus sur la récupération des chaîne de connexion avec des principaux de sécurité qui ne sont pas accordés avec Data des rôles associés, consultez la section Utilisation de base de Spring Could Azure Resource Manager.

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-stream-binder-eventhubs :

Propriétés de configuration de liaison Azure Event Hubs

Les options suivantes sont divisées en quatre sections : Propriétés du consommateur, Configurations avancées du consommateur, Propriétés du producteur et Configurations de producteur avancées.

Propriétés du consommateur

Ces propriétés sont exposées via EventHubsConsumerProperties.

Propriétés configurables par le consommateur de spring-cloud-azure-stream-binder-eventhubs :

spring:
  cloud:
    stream:
      eventhubs:
        bindings:
          <binding-name>:
            consumer:
              initial-partition-event-position:
                0:
                  offset: earliest
                1:
                  sequence-number: 100
                2:
                  enqueued-date-time: 2022-01-12T13:32:47.650005Z
                4:
                  inclusive: false

Configuration avancée du consommateur

La connexion ci-dessus, case activée point et la configuration du client du Kit de développement logiciel (SDK) Azure courante prennent en charge la personnalisation de chaque consommateur de classeurs, que vous pouvez configurer avec le préfixe spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer..

Propriétés du producteur

Ces propriétés sont exposées via EventHubsProducerProperties.

Propriétés configurables par le producteur de spring-cloud-azure-stream-binder-eventhubs :

Configuration avancée du producteur

La connexion ci-dessus et la configuration du client du Kit de développement logiciel (SDK) Azure courante prennent en charge la personnalisation de chaque producteur de classeurs, que vous pouvez configurer avec le préfixe spring.cloud.stream.eventhubs.bindings.<binding-name>.producer..

Utilisation de base

Envoi et réception de messages depuis/vers Event Hubs

1. Renseignez les options de configuration avec 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: ${EVENTHUB_NAMESPACE_CONNECTION_STRING}
             processor:
               checkpoint-store:
                 container-name: ${CHECKPOINT_CONTAINER}
                 account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                 account-key: ${CHECKPOINT_ACCESS_KEY}
         function:
           definition: consume;supply
         stream:
           bindings:
             consume-in-0:
               destination: ${EVENTHUB_NAME}
               group: ${CONSUMER_GROUP}
             supply-out-0:
               destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
           eventhubs:
             bindings:
               consume-in-0:
                 consumer:
                   checkpoint:
                     mode: MANUAL
     

   • 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: ${EVENTHUB_NAMESPACE}
             processor:
               checkpoint-store:
                 container-name: ${CONTAINER_NAME}
                 account-name: ${ACCOUNT_NAME}
         function:
           definition: consume;supply
         stream:
           bindings:
             consume-in-0:
               destination: ${EVENTHUB_NAME}
               group: ${CONSUMER_GROUP}
             supply-out-0:
               destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
           eventhubs:
             bindings:
               consume-in-0:
                 consumer:
                   checkpoint:
                     mode: MANUAL
     

• 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_MANAGED_IDENTITY_CLIENT_ID} # Only needed when using a user-assigned managed identity
        eventhubs:
          namespace: ${EVENTHUB_NAMESPACE}
          processor:
            checkpoint-store:
              container-name: ${CONTAINER_NAME}
              account-name: ${ACCOUNT_NAME}
      function:
        definition: consume;supply
      stream:
        bindings:
          consume-in-0:
            destination: ${EVENTHUB_NAME}
            group: ${CONSUMER_GROUP}
          supply-out-0:
            destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
  
        eventhubs:
          bindings:
            consume-in-0:
              consumer:
                checkpoint:
                  mode: MANUAL
  

1. Définir le fournisseur et le consommateur.

   @Bean
   public Consumer<Message<String>> consume() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                   message.getPayload(),
                   message.getHeaders().get(EventHubsHeaders.PARTITION_KEY),
                   message.getHeaders().get(EventHubsHeaders.SEQUENCE_NUMBER),
                   message.getHeaders().get(EventHubsHeaders.OFFSET),
                   message.getHeaders().get(EventHubsHeaders.ENQUEUED_TIME)
           );
   
           checkpointer.success()
                   .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("Hello world, " + i++).build();
       };
   }
   

Prise en charge du partitionnement

Une PartitionSupplier information de partition fournie par l’utilisateur est créée pour configurer les informations de partition sur le message à envoyer. L’organigramme suivant montre le processus d’obtention de priorités différentes pour l’ID de partition et la clé :

Prise en charge des consommateurs Batch

1. Fournissez les options de configuration par lots, comme indiqué dans l’exemple suivant :

   spring:
     cloud:
       function:
         definition: consume
       stream:
         bindings:
           consume-in-0:
             destination: ${AZURE_EVENTHUB_NAME}
             group: ${AZURE_EVENTHUB_CONSUMER_GROUP}
             consumer:
               batch-mode: true
         eventhubs:
           bindings:
             consume-in-0:
               consumer:
                 batch:
                   max-batch-size: 10 # Required for batch-consumer mode
                   max-wait-time: 1m # Optional, the default value is null
                 checkpoint:
                   mode: BATCH # or MANUAL as needed
   

2. Définir le fournisseur et le consommateur.

   Pour case activée mode de pointage, vous BATCHpouvez utiliser le code suivant pour envoyer des messages et les consommer par lots.

   @Bean
   public Consumer<Message<List<String>>> consume() {
       return message -> {
           for (int i = 0; i < message.getPayload().size(); i++) {
               LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                       message.getPayload().get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_PARTITION_KEY)).get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_SEQUENCE_NUMBER)).get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_OFFSET)).get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_ENQUEUED_TIME)).get(i));
           }
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("\"test"+ i++ +"\"").build();
       };
   }
   

   Pour case activée mode de pointage, vous MANUALpouvez utiliser le code suivant pour envoyer des messages et consommer/case activée point par lots.

   @Bean
   public Consumer<Message<List<String>>> consume() {
       return message -> {
           for (int i = 0; i < message.getPayload().size(); i++) {
               LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                   message.getPayload().get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_PARTITION_KEY)).get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_SEQUENCE_NUMBER)).get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_OFFSET)).get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_ENQUEUED_TIME)).get(i));
           }
   
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           checkpointer.success()
                       .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                       .doOnError(error -> LOGGER.error("Exception found", error))
                       .block();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("\"test"+ i++ +"\"").build();
       };
   }
   

Traiter les messages d’erreur

En-têtes de message Event Hubs

Pour connaître les en-têtes de message de base pris en charge, consultez la section En-têtes de message Event Hubs de Spring Cloud support Azure pour Spring Integration.

Prise en charge de plusieurs classeurs

Connecter ion à plusieurs espaces de noms Event Hubs est également prise en charge à l’aide de plusieurs classeurs. Cet exemple prend un chaîne de connexion comme exemple. Les informations d’identification des principaux de service et des identités managées sont également prises en charge. Vous pouvez définir des propriétés associées dans les paramètres d’environnement de chaque classeur.

1. Pour utiliser plusieurs classeurs avec Event Hubs, configurez les propriétés suivantes dans votre fichier application.yml :

   spring:
     cloud:
       function:
         definition: consume1;supply1;consume2;supply2
       stream:
         bindings:
           consume1-in-0:
             destination: ${EVENTHUB_NAME_01}
             group: ${CONSUMER_GROUP_01}
           supply1-out-0:
             destination: ${THE_SAME_EVENTHUB_NAME_01_AS_ABOVE}
           consume2-in-0:
             binder: eventhub-2
             destination: ${EVENTHUB_NAME_02}
             group: ${CONSUMER_GROUP_02}
           supply2-out-0:
             binder: eventhub-2
             destination: ${THE_SAME_EVENTHUB_NAME_02_AS_ABOVE}
         binders:
           eventhub-1:
             type: eventhubs
             default-candidate: true
             environment:
               spring:
                 cloud:
                   azure:
                     eventhubs:
                       connection-string: ${EVENTHUB_NAMESPACE_01_CONNECTION_STRING}
                       processor:
                         checkpoint-store:
                           container-name: ${CHECKPOINT_CONTAINER_01}
                           account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                           account-key: ${CHECKPOINT_ACCESS_KEY}
           eventhub-2:
             type: eventhubs
             default-candidate: false
             environment:
               spring:
                 cloud:
                   azure:
                     eventhubs:
                       connection-string: ${EVENTHUB_NAMESPACE_02_CONNECTION_STRING}
                       processor:
                         checkpoint-store:
                           container-name: ${CHECKPOINT_CONTAINER_02}
                           account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                           account-key: ${CHECKPOINT_ACCESS_KEY}
         eventhubs:
           bindings:
             consume1-in-0:
               consumer:
                 checkpoint:
                   mode: MANUAL
             consume2-in-0:
               consumer:
                 checkpoint:
                   mode: MANUAL
         poller:
           initial-delay: 0
           fixed-delay: 1000
   

   Remarque

   Le fichier d’application précédent montre comment configurer un seul polleur par défaut pour l’application sur toutes les liaisons. Si vous souhaitez configurer l’polleur pour une liaison spécifique, vous pouvez utiliser une configuration telle que spring.cloud.stream.bindings.<binding-name>.producer.poller.fixed-delay=3000.

2. Nous devons définir deux fournisseurs et deux consommateurs :

   @Bean
   public Supplier<Message<String>> supply1() {
       return () -> {
           LOGGER.info("Sending message1, sequence1 " + i);
           return MessageBuilder.withPayload("Hello world1, " + i++).build();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply2() {
       return () -> {
           LOGGER.info("Sending message2, sequence2 " + j);
           return MessageBuilder.withPayload("Hello world2, " + j++).build();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume1() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message1 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(success -> LOGGER.info("Message1 '{}' successfully checkpointed", message))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume2() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message2 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(success -> LOGGER.info("Message2 '{}' successfully checkpointed", message))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   

Provisionnement des ressources

Event Hubs binder prend en charge l’approvisionnement d’event hub et de groupe de consommateurs, les utilisateurs peuvent utiliser les propriétés suivantes pour activer l’approvisionnement.

spring:
  cloud:
    azure:
      credential:
        tenant-id: <tenant>
      profile:
        subscription-id: ${AZURE_SUBSCRIPTION_ID}
      eventhubs:
        resource:
          resource-group: ${AZURE_EVENTHUBS_RESOURECE_GROUP}

Exemples

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

Spring Cloud Stream Binder pour Azure Service Bus

Concepts clés

Spring Cloud Stream Binder pour Azure Service Bus fournit l’implémentation de liaison pour Spring Cloud Stream Framework. Cette implémentation utilise les adaptateurs de canal Spring Integration Service Bus à sa base.

Message planifié

Ce classeur prend en charge l’envoi de messages à une rubrique pour le traitement différé. Les utilisateurs peuvent envoyer des messages planifiés avec l’en-tête x-delay exprimant en millisecondes un délai pour le message. Le message est remis aux rubriques respectives après x-delay millisecondes.

Groupe de consommateurs

Service Bus Topic fournit une prise en charge similaire du groupe de consommateurs comme Apache Kafka, mais avec une légère logique différente. Ce classeur s’appuie sur Subscription un sujet pour agir en tant que groupe de consommateurs.

Configuration des dépendances

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-stream-binder-servicebus</artifactId>
</dependency>

Vous pouvez également utiliser Spring Cloud Azure Stream Service Bus Starter, comme illustré dans l’exemple suivant pour Maven :

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

Configuration

Le classeur fournit les deux 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-stream-binder-servicebus :

Le classeur prend également en charge Spring Could Azure Resource Manager par défaut. Pour en savoir plus sur la récupération des chaîne de connexion avec des principaux de sécurité qui ne sont pas accordés avec Data des rôles associés, consultez la section Utilisation de base de Spring Could Azure Resource Manager.

Propriétés de configuration de la liaison Azure Service Bus

Les options suivantes sont divisées en quatre sections : Propriétés du consommateur, Configurations avancées du consommateur, Propriétés du producteur et Configurations de producteur avancées.

Propriétés du consommateur

Ces propriétés sont exposées via ServiceBusConsumerProperties.

Propriétés configurables par le consommateur de spring-cloud-azure-stream-binder-servicebus :

Configuration avancée du consommateur

La connexion ci-dessus et la configuration du client du Kit de développement logiciel (SDK) Azure courante prennent en charge la personnalisation pour chaque consommateur de classeurs, que vous pouvez configurer avec le préfixe spring.cloud.stream.servicebus.bindings.<binding-name>.consumer..

Propriétés du producteur

Ces propriétés sont exposées via ServiceBusProducerProperties.

Propriétés configurables par le producteur de spring-cloud-azure-stream-binder-servicebus :

Configuration avancée du producteur

La connexion ci-dessus et la configuration du client du Kit de développement logiciel (SDK) Azure courante prennent en charge la personnalisation de chaque producteur de classeurs, que vous pouvez configurer avec le préfixe spring.cloud.stream.servicebus.bindings.<binding-name>.producer..

Utilisation de base

Envoi et réception de messages depuis/vers Service Bus

1. Renseignez les options de configuration avec 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: ${SERVICEBUS_NAMESPACE_CONNECTION_STRING}
             function:
               definition: consume;supply
             stream:
               bindings:
                 consume-in-0:
                   destination: ${SERVICEBUS_ENTITY_NAME}
                   # If you use Service Bus Topic, add the following configuration
                   # group: ${SUBSCRIPTION_NAME}
                 supply-out-0:
                   destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
               servicebus:
                 bindings:
                   consume-in-0:
                     consumer:
                       auto-complete: false
                   supply-out-0:
                     producer:
                       entity-type: queue # set as "topic" if you use Service Bus Topic
     

   • 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: ${SERVICEBUS_NAMESPACE}
             function:
               definition: consume;supply
             stream:
               bindings:
                 consume-in-0:
                   destination: ${SERVICEBUS_ENTITY_NAME}
                   # If you use Service Bus Topic, add the following configuration
                   # group: ${SUBSCRIPTION_NAME}
                 supply-out-0:
                   destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
               servicebus:
                 bindings:
                   consume-in-0:
                     consumer:
                       auto-complete: false
                   supply-out-0:
                     producer:
                       entity-type: queue # set as "topic" if you use Service Bus Topic
     

• 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: ${MANAGED_IDENTITY_CLIENT_ID} # Only needed when using a user-assigned managed identity
            servicebus:
              namespace: ${SERVICEBUS_NAMESPACE}
          function:
            definition: consume;supply
          stream:
            bindings:
              consume-in-0:
                destination: ${SERVICEBUS_ENTITY_NAME}
                # If you use Service Bus Topic, add the following configuration
                # group: ${SUBSCRIPTION_NAME}
              supply-out-0:
                destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
            servicebus:
              bindings:
                consume-in-0:
                  consumer:
                    auto-complete: false
                supply-out-0:
                  producer:
                    entity-type: queue # set as "topic" if you use Service Bus Topic
  

1. Définir le fournisseur et le consommateur.

   @Bean
   public Consumer<Message<String>> consume() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message received: '{}'", message.getPayload());
   
           checkpointer.success()
                   .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("Hello world, " + i++).build();
       };
   }
   

Prise en charge des clés de partition

Le classeur 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.

Spring Cloud Stream fournit une propriété spring.cloud.stream.bindings.<binding-name>.producer.partition-key-expressiond’expression SpEL de clé de partition. Par exemple, la définition de cette propriété et "'partitionKey-' + headers[<message-header-key>]" l’ajout d’un en-tête appelé message-header-key. Spring Cloud Stream utilise la valeur de cet en-tête lors de l’évaluation de l’expression pour affecter une clé de partition. Le code suivant fournit un exemple de producteur :

@Bean
public Supplier<Message<String>> generate() {
    return () -> {
        String value = "random payload";
        return MessageBuilder.withPayload(value)
            .setHeader("<message-header-key>", value.length() % 4)
            .build();
    };
}

Prise en charge des sessions

Le classeur prend en charge les sessions de message de Service Bus. L’ID de session d’un message peut être défini via l’en-tête du message.

@Bean
public Supplier<Message<String>> generate() {
    return () -> {
        String value = "random payload";
        return MessageBuilder.withPayload(value)
            .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session ID")
            .build();
    };
}

Traiter les messages d’erreur

En-têtes de message Service Bus

Pour connaître les en-têtes de message de base pris en charge, consultez la section En-têtes de message Service Bus de Spring Cloud support Azure pour Spring Integration.

Prise en charge de plusieurs classeurs

Connecter ion à plusieurs espaces de noms Service Bus est également prise en charge à l’aide de plusieurs classeurs. Cet exemple prend chaîne de connexion comme exemple. Les informations d’identification des principaux de service et des identités managées sont également prises en charge, les utilisateurs peuvent définir des propriétés associées dans les paramètres d’environnement de chaque classeur.

1. Pour utiliser plusieurs classeurs de ServiceBus, configurez les propriétés suivantes dans votre fichier application.yml :

   spring:
     cloud:
       function:
         definition: consume1;supply1;consume2;supply2
       stream:
         bindings:
           consume1-in-0:
             destination: ${SERVICEBUS_TOPIC_NAME}
             group: ${SUBSCRIPTION_NAME}
           supply1-out-0:
             destination: ${SERVICEBUS_TOPIC_NAME_SAME_AS_ABOVE}
           consume2-in-0:
             binder: servicebus-2
             destination: ${SERVICEBUS_QUEUE_NAME}
           supply2-out-0:
             binder: servicebus-2
             destination: ${SERVICEBUS_QUEUE_NAME_SAME_AS_ABOVE}
         binders:
           servicebus-1:
             type: servicebus
             default-candidate: true
             environment:
               spring:
                 cloud:
                   azure:
                     servicebus:
                       connection-string: ${SERVICEBUS_NAMESPACE_01_CONNECTION_STRING}
           servicebus-2:
             type: servicebus
             default-candidate: false
             environment:
               spring:
                 cloud:
                   azure:
                     servicebus:
                       connection-string: ${SERVICEBUS_NAMESPACE_02_CONNECTION_STRING}
         servicebus:
           bindings:
             consume1-in-0:
               consumer:
                 auto-complete: false
             supply1-out-0:
               producer:
                 entity-type: topic
             consume2-in-0:
               consumer:
                 auto-complete: false
             supply2-out-0:
               producer:
                 entity-type: queue
         poller:
           initial-delay: 0
           fixed-delay: 1000
   

   Remarque

   Le fichier d’application précédent montre comment configurer un seul polleur par défaut pour l’application sur toutes les liaisons. Si vous souhaitez configurer l’polleur pour une liaison spécifique, vous pouvez utiliser une configuration telle que spring.cloud.stream.bindings.<binding-name>.producer.poller.fixed-delay=3000.

2. nous devons définir deux fournisseurs et deux consommateurs

   @Bean
   public Supplier<Message<String>> supply1() {
       return () -> {
           LOGGER.info("Sending message1, sequence1 " + i);
           return MessageBuilder.withPayload("Hello world1, " + i++).build();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply2() {
       return () -> {
           LOGGER.info("Sending message2, sequence2 " + j);
           return MessageBuilder.withPayload("Hello world2, " + j++).build();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume1() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message1 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(e -> LOGGER.error("Error found", e))
                   .block();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume2() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message2 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(e -> LOGGER.error("Error found", e))
                   .block();
       };
   
   }
   

Provisionnement des ressources

Service Bus binder prend en charge l’approvisionnement de files d’attente, de rubriques et d’abonnements, les utilisateurs peuvent utiliser les propriétés suivantes pour activer l’approvisionnement.

spring:
  cloud:
    azure:
      credential:
        tenant-id: <tenant>
      profile:
        subscription-id: ${AZURE_SUBSCRIPTION_ID}
      servicebus:
        resource:
          resource-group: ${AZURE_SERVICEBUS_RESOURECE_GROUP}
    stream:
      servicebus:
        bindings:
          <binding-name>:
            consumer:
              entity-type: ${SERVICEBUS_CONSUMER_ENTITY_TYPE}

Exemples

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