Partager via

Spring Cloud Stream avec Azure Event Hubs

  • Article

Ce guide explique comment envoyer et recevoir des messages en utilisant Azure Event Hubs et Spring Cloud Stream Binder Eventhubs dans une application Spring Boot.

Prérequis

Remarque

Pour accorder à votre compte l'accès aux ressources, dans Azure Event Hubs, attribuez le rôle Azure Event Hubs Data Receiver et Azure Event Hubs Data Sender au compte Microsoft Entra que vous utilisez actuellement. Ensuite, dans le compte Azure Storage, attribuez le rôle Storage Blob Data Contributor au compte Microsoft Entra que vous utilisez actuellement. Pour plus d'informations sur l'attribution de rôles d'accès, consultez les sections Attribuer des rôles Azure à l'aide du portail Azure et Autoriser l'accès aux ressources Event Hubs à l'aide de Microsoft Entra ID.

Important

La version 2.5 ou supérieure de Spring Boot est nécessaire pour réaliser les étapes de ce tutoriel.

Envoyer et recevoir des messages depuis des hubs d'événements Azure

Avec un compte de stockage Azure et un hub d'événements Azure, vous pouvez envoyer et recevoir des messages via les hubs d'événements à l'aide de Spring Cloud Azure Stream Binder.

Pour installer le module Spring Cloud Azure Stream Binder Event Hubs, ajoutez les dépendances suivantes à votre fichier pom.xml :

  • La nomenclature Spring Cloud Azure :

    <dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>com.azure.spring</groupId>
      <artifactId>spring-cloud-azure-dependencies</artifactId>
      <version>5.18.0</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

    Remarque

    Si vous utilisez Spring Boot 2.x, assurez-vous de définir la version spring-cloud-azure-dependencies sur 4.19.0. Cette nomenclature doit être configurée dans la section <dependencyManagement> de votre fichier pom.xml. Cela permet de s'assurer que toutes les dépendances de Spring Cloud Azure utilisent la même version. Pour plus d'informations sur la version utilisée pour cette nomenclature, consultez Quelle version de Spring Cloud Azure dois-je utiliser.

  • L'artefact Spring Cloud Azure Stream Binder Event Hubs :

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

Coder l’application

Suivez les étapes ci-dessous pour configurer votre application afin qu'elle produise et consomme des messages via Azure Event Hubs.

  1. Configurez les informations d'identification du hub d'événements en ajoutant les propriétés suivantes à votre fichier application.properties.

     spring.cloud.azure.eventhubs.namespace=${AZURE_EVENTHUBS_NAMESPACE}
 spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=${AZURE_STORAGE_ACCOUNT_NAME}
 spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=${AZURE_STORAGE_CONTAINER_NAME}
 spring.cloud.stream.bindings.consume-in-0.destination=${AZURE_EVENTHUB_NAME}
 spring.cloud.stream.bindings.consume-in-0.group=${AZURE_EVENTHUB_CONSUMER_GROUP}
 spring.cloud.stream.bindings.supply-out-0.destination=${AZURE_EVENTHUB_NAME}
 spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode=MANUAL
 spring.cloud.function.definition=consume;supply;
 spring.cloud.stream.poller.initial-delay=0
 spring.cloud.stream.poller.fixed-delay=1000

    Le tableau suivant décrit les champs de la configuration :

    Champ Description
    spring.cloud.azure.eventhubs.namespace Indiquez l'espace de noms que vous avez obtenu pour votre hub d'événements depuis le portail Azure.
    spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name Spécifiez le compte de stockage que vous avez créé dans ce tutoriel.
    spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name Spécifiez le conteneur de votre compte de stockage.
    spring.cloud.stream.bindings.consume-in-0.destination Indiquez le hub d'événements utilisé dans ce guide.
    spring.cloud.stream.bindings.consume-in-0.group Spécifiez les groupes de consommateurs de votre instance Event Hubs.
    spring.cloud.stream.bindings.supply-out-0.destination Indiquez le même hub d'événements que celui utilisé dans ce guide.
    spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode Spécifier MANUAL.
    spring.cloud.function.definition Spécifiez le composant Bean fonctionnel à lier aux destinations externes qui sont exposées par les liaisons.
    spring.cloud.stream.poller.initial-delay Spécifiez le délai initial pour les déclencheurs périodiques. La valeur par défaut est 0.
    spring.cloud.stream.poller.fixed-delay Spécifiez le délai fixe pour le poller par défaut en millisecondes. La valeur par défaut est de 1000 L.

  2. Modifiez le fichier de classe de démarrage pour afficher le contenu suivant.

    import com.azure.spring.messaging.checkpoint.Checkpointer;
import com.azure.spring.messaging.eventhubs.support.EventHubsHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.util.function.Consumer;
import java.util.function.Supplier;
import static com.azure.spring.messaging.AzureHeaders.CHECKPOINTER;

@SpringBootApplication
public class EventHubBinderApplication implements CommandLineRunner {

    private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderApplication.class);
    private static final Sinks.Many<Message<String>> many = Sinks.many().unicast().onBackpressureBuffer();

    public static void main(String[] args) {
        SpringApplication.run(EventHubBinderApplication.class, args);
    }

    @Bean
    public Supplier<Flux<Message<String>>> supply() {
        return ()->many.asFlux()
                       .doOnNext(m->LOGGER.info("Manually sending message {}", m))
                       .doOnError(t->LOGGER.error("Error encountered", t));
    }

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

    @Override
    public void run(String... args) {
        LOGGER.info("Going to add message {} to sendMessage.", "Hello World");
        many.emitNext(MessageBuilder.withPayload("Hello World").build(), Sinks.EmitFailureHandler.FAIL_FAST);
    }

}

    Conseil

    Dans ce tutoriel, il n'y a pas d'opérations d'authentification dans les configurations ou le code. Cependant, la connexion aux services Azure nécessite une authentification. Pour effectuer l’authentification, vous devez utiliser Identité Azure. Spring Cloud Azure utilise DefaultAzureCredential, que la bibliothèque Azure Identity fournit pour vous aider à obtenir des informations d'identification sans modifier le code.

    DefaultAzureCredential prend en charge plusieurs méthodes d’authentification et détermine quelle méthode doit être utilisée au moment de l’exécution. Cette approche permet à votre application d'utiliser différentes méthodes d'authentification dans différents environnements (tels que les environnements locaux et de production) sans implémenter de code spécifique à l'environnement. Pour plus d’informations, consultez DefaultAzureCredential.

    Pour réaliser l'authentification dans les environnements de développement locaux, vous pouvez utiliser Azure CLI, Visual Studio Code, PowerShell ou d'autres méthodes. Pour plus d'informations, consultez la section Authentification Azure dans les environnements de développement Java. Pour compléter l'authentification dans les environnements d'hébergement Azure, nous vous recommandons d'utiliser l'identité gérée attribuée à l'utilisateur. Pour plus d’informations, consultez Que sont les identités managées pour les ressources Azure ?

  3. Lancez l’application. Les messages de ce type apparaîtront dans le journal de votre application, comme illustré ci-dessous :

    New message received: 'Hello World', partition key: 107207233, sequence number: 458, offset: 94256, enqueued time: 2023-02-17T08:27:59.641Z
Message 'Hello World!' successfully checkpointed

Déployer sur Azure Spring Apps

Maintenant que l'application Spring Boot fonctionne localement, il est temps de la mettre en production. Azure Spring Apps facilite le déploiement des applications Spring Boot sur Azure sans aucune modification du code. Le service gère l’infrastructure des applications Spring, ce qui permet aux développeurs de se concentrer sur leur code. Azure Spring Apps assure la gestion du cycle de vie en utilisant des outils complets, tels que la supervision et les diagnostics, la gestion des configurations, la découverte de services, l’intégration CI/CD, les déploiements bleus-verts, etc. Pour déployer votre application sur Azure Spring Apps, voir Déployer votre première application sur Azure Spring Apps.

Étapes suivantes

Azure pour les développeurs SpringSpring Cloud, Azure Stream Binder Event & échantillons de Hubs