Använda Event Hubs i Spring-program

Den här artikeln visar hur du använder Azure Event Hubs i Java program som skapats med Spring Framework.

Azure Event Hubs är en stordataströmningsplattform och händelseinmatningstjänst. Den kan ta emot och behandla miljoner händelser per sekund. Data som skickas till en händelsehubb kan omvandlas och lagras med hjälp av valfri provider för realtidsanalys eller batchbearbetnings-/lagringsadaptrar.

Spring Cloud Azure innehåller olika moduler för att skicka meddelanden till och ta emot meddelanden från Event Hubs med springramverk.

Du kan använda följande moduler oberoende av varandra eller kombinera dem för olika användningsfall:

• Spring Cloud Azure Event Hubs Starter kan du skicka och ta emot meddelanden med Event Hubs Java SDK-klientbibliotek med Spring Boot-funktioner.

• Spring Messaging Azure Event Hubs gör att du kan interagera med Event Hubs via Spring Messaging API.

• Spring Integration Azure Event Hubs kan du ansluta Spring Integration Message Channels med Event Hubs.

• Spring Cloud Azure Stream Event Hubs Binder gör att du kan använda Event Hubs som mellanprogram för meddelanden i Spring Cloud Stream-program.

• Spring Kafka med Azure Event Hubs kan du använda Spring Kafka för att skicka meddelanden till och ta emot meddelanden från Event Hubs.

• Spring Cloud Stream Kafka Binder med Azure Event Hubs gör att du kan skicka och ta emot meddelanden via Spring Cloud Stream Kafka Binder med Event Hubs.

Förutsättningar

• En Azure prenumeration – skapa en kostnadsfri.

• Java Development Kit (JDK) version 8 eller senare.

• Apache Maven.

• Azure CLI.

• En "Azure Event Hubs"-instans. Mer information finns i Quickstart: Skapa en händelsehubb med Azure portal.

• Ett Azure Storage-konto för Event Hubs-kontrollpunkter. Mer information finns i Skapa ett lagringskonto.

• Ett Spring Boot-program. Om du inte har ett skapar du ett Maven-projekt med Spring Initializr. Kom ihåg att välja Maven Project och under Dependencies lägger du till Spring Web beroende och väljer sedan Java version 8 eller senare.

Anteckning

Om du vill ge ditt konto åtkomst till resurser i Azure Event Hubs tilldelar du rollen Azure Event Hubs Data Receiver och Azure Event Hubs Data Sender till det Microsoft Entra konto som du använder för närvarande. I Azure Storage-kontot tilldelar du sedan rollen Storage Blob Data Contributor till det Microsoft Entra konto som du använder för närvarande. Mer information om hur du beviljar åtkomstroller finns i Assign Azure roles using the Azure portal and Authorize access to Event Hubs resources using Microsoft Entra ID.

Viktigt!

Spring Boot version 2.5 eller senare krävs för att slutföra stegen i den här självstudien.

Förbereda din lokala miljö

I den här handledningen finns det inga autentiseringsåtgärder i konfigurationerna och koden. Att ansluta till en Azure tjänst kräver dock autentisering. För att slutföra autentiseringen måste du använda klientbiblioteket Azure Identity. Spring Cloud Azure använder DefaultAzureCredential, som Azure Identity-biblioteket tillhandahåller för att hjälpa dig att få autentiseringsuppgifter utan några kodändringar.

DefaultAzureCredential stöder flera autentiseringsmetoder och avgör vilken metod som ska användas vid körning. Med den här metoden kan din app använda olika autentiseringsmetoder i olika miljöer , till exempel lokala miljöer eller produktionsmiljöer, utan att implementera miljöspecifik kod. Mer information finns i avsnittet DefaultAzureCredential i Authenticate Azure-hosted Java applications.

Om du vill använda Azure CLI, IntelliJ eller andra metoder för att slutföra autentiseringen i lokala utvecklingsmiljöer kan du läsa Azure autentisering i Java utvecklingsmiljöer. För att slutföra autentiseringen i Azure värdmiljöer rekommenderar vi att du använder hanterad identitet. Mer information finns i Vad är hanterade identiteter för Azure resurser?

Använda Spring Cloud Azure Event Hubs Starter

Modulen Spring Cloud Azure Event Hubs Starter importerar Event Hubs Java klientbibliotek med Spring Boot-ramverket. Du kan använda Spring Cloud Azure och Azure SDKs tillsammans i ett icke-ömsesidigt uteslutande mönster. Därför kan du fortsätta använda Event Hubs Java-klient-API:et i ditt Spring-program.

Lägga till beroenden

Om du vill installera Modulen Spring Cloud Azure Event Hubs Starter lägger du till följande beroenden i filen pom.xml:

• Spring Cloud Azure Materialförteckning (BOM):

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

  Anteckning

  Om du använder Spring Boot 4.0.x anger du spring-cloud-azure-dependencies versionen till 7.4.0.

  Om du använder Spring Boot 3.5.x anger du spring-cloud-azure-dependencies versionen till 6.5.0.

  Om du använder Spring Boot 3.1.x-3.5.x måste du ange spring-cloud-azure-dependencies versionen till 5.25.0.

  Om du använder Spring Boot 2.x måste du ange spring-cloud-azure-dependencies versionen till 4.20.0.

  Den här materialspecifikationen (BOM) bör konfigureras i avsnittet <dependencyManagement> i din pom.xml-fil. Detta säkerställer att alla Spring Cloud-Azure beroenden använder samma version.

  Mer information om vilken version som används för den här BOM finns genom att läsa Which Version of Spring Cloud Azure Should I Use.

• Spring Cloud Azure Event Hubs artefakt:

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

Koda programmet för att skicka och ta emot meddelanden

Den här guiden lär dig hur du använder Event Hubs-Java klienter i samband med ett Spring-program. Här introducerar vi följande två alternativ:

• Använd Autokonfiguration av Spring Boot och använd färdiga klienter från Spring-kontexten (rekommenderas).
• Skapa klienten programmatiskt.

Metoden för att automatiskt koppla klientkomponenter i Spring IoC-containern har följande fördelar, vilket kan ge dig en mer flexibel och effektiv upplevelse när du utvecklar med Event Hubs-klienter.

• Den tillämpar den externaliserade konfigurationen så att du kan arbeta med samma programkod i olika miljöer.
• Du kan delegera processen att lära sig byggmönstret och registrera klienten i programkontexten till Spring Boot-ramverket. Med den här delegeringen kan du fokusera på hur du använder klienterna med dina egna affärsbehov.
• Du kan använda hälsoindikatorn på ett enkelt sätt för att inspektera status och hälsa för ditt program och dina interna komponenter.

Följande avsnitt innehåller kodexempel som visar hur du använder EventProcessorClient och EventHubProducerClient med de två alternativen.

Anteckning

Azure Java SDK för Event Hubs ger flera klienter att interagera med Event Hubs. Startprogrammet tillhandahåller även automatisk konfiguration för alla Event Hubs-klienter och klientbyggare. Den här artikeln använder endast EventProcessorClient och EventHubProducerClient som exempel.

Använda Autokonfiguration av Spring Boot

Om du vill skicka meddelanden till och ta emot meddelanden från Event Hubs konfigurerar du programmet med hjälp av följande steg:

1. Använd följande egenskapsinställningar för att konfigurera Event Hubs-namnrymden och händelsehubbens namn:

   spring.cloud.azure.eventhubs.namespace=<your event-hubs-namespace>
   spring.cloud.azure.eventhubs.event-hub-name=<your-event-hub-name>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=<your-storage-account-name>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=<your-storage-account-container-name>
   spring.cloud.azure.eventhubs.processor.consumer-group=$Default
   

2. Skapa en ny klass EventHubProcessorClientConfiguration Java enligt följande exempel. Den här klassen används för att registrera meddelandet och felhanteraren för EventProcessorClient.

   import com.azure.spring.cloud.service.eventhubs.consumer.EventHubsErrorHandler;
   import com.azure.spring.cloud.service.eventhubs.consumer.EventHubsRecordMessageListener;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.context.annotation.Bean;
   import org.springframework.context.annotation.Configuration;
   
   @Configuration
   public class EventHubProcessorClientConfiguration {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubProcessorClientConfiguration.class);
   
       @Bean
       EventHubsRecordMessageListener processEvent() {
           return eventContext->LOGGER.info("Processing event from partition {} with sequence number {} with body: {}",
               eventContext.getPartitionContext().getPartitionId(), eventContext.getEventData().getSequenceNumber(),
               eventContext.getEventData().getBodyAsString());
       }
   
       @Bean
       EventHubsErrorHandler processError() {
           return errorContext->LOGGER.info("Error occurred in partition processor for partition {}, {}",
               errorContext.getPartitionContext().getPartitionId(),
               errorContext.getThrowable());
       }
   
   }
   

3. EventProcessorClient Mata in och EventHubProducerClient i Spring-programmet och anropa de relaterade API:erna för att skicka och ta emot meddelanden, som du ser i följande exempel:

   import com.azure.messaging.eventhubs.EventData;
   import com.azure.messaging.eventhubs.EventHubProducerClient;
   import com.azure.messaging.eventhubs.EventProcessorClient;
   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 java.util.Collections;
   import java.util.concurrent.TimeUnit;
   
   @SpringBootApplication
   public class EventHubClientApplication implements CommandLineRunner {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubClientApplication.class);
       private final EventHubProducerClient eventHubProducerClient;
       private final EventProcessorClient eventProcessorClient;
   
       public EventHubClientApplication(EventHubProducerClient eventHubProducerClient,
                                        EventProcessorClient eventProcessorClient) {
           this.eventHubProducerClient = eventHubProducerClient;
           this.eventProcessorClient = eventProcessorClient;
       }
   
       public static void main(String[] args) {
           SpringApplication.run(EventHubClientApplication.class, args);
       }
   
       @Override
       public void run(String... args) throws Exception {
           eventProcessorClient.start();
           // Wait for the processor client to be ready
           TimeUnit.SECONDS.sleep(10);
   
           eventHubProducerClient.send(Collections.singletonList(new EventData("Hello World")));
           LOGGER.info("Successfully sent a message to Event Hubs.");
           eventHubProducerClient.close();
           LOGGER.info("Skip stopping and closing the processor since the processor may not complete the receiving process yet.");
       }
   
   }
   

4. Starta programmet. Du visas loggar som liknar följande exempel:

   Successfully sent a message to Event Hubs.
   ...
   Processing event from partition 0 with sequence number 0 with body: Hello World
   ...
   Stopping and closing the processor.
   

Skapa klienten programmatiskt

Du kan skapa klientkomponenterna själv, men processen är komplicerad. I Spring Boot-program måste du hantera egenskaper, lära dig builder-mönstret och registrera klienten i din Spring-programkontext. Följande steg visar hur du gör det:

1. Skapa en ny klass EventHubClientConfiguration Java enligt följande exempel. Den här klassen används för att deklarera bönorna EventProcessorClient och EventHubProducerClient . Se till att ersätta <your event-hubs-namespace>platshållarna , <your-event-hub-name>, <your-storage-account-name>och <your-storage-account-container-name> med dina faktiska värden.

   import com.azure.identity.DefaultAzureCredentialBuilder;
   import com.azure.messaging.eventhubs.EventHubClientBuilder;
   import com.azure.messaging.eventhubs.EventHubProducerClient;
   import com.azure.messaging.eventhubs.EventProcessorClient;
   import com.azure.messaging.eventhubs.EventProcessorClientBuilder;
   import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
   import com.azure.messaging.eventhubs.models.ErrorContext;
   import com.azure.messaging.eventhubs.models.EventContext;
   import com.azure.storage.blob.BlobContainerAsyncClient;
   import com.azure.storage.blob.BlobContainerClientBuilder;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.context.annotation.Bean;
   import org.springframework.context.annotation.Configuration;
   
   @Configuration
   public class EventHubClientConfiguration {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubClientConfiguration.class);
       private static final String EVENT_HUB_FULLY_QUALIFIED_NAMESPACE = "<your event-hubs-namespace>.servicebus.windows.net";
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
       private static final String CONSUMER_GROUP = "$Default";
       private static final String STORAGE_ACCOUNT_ENDPOINT = "https://<your-storage-account-name>.blob.core.windows.net";
       private static final String STORAGE_CONTAINER_NAME = "<your-storage-account-container-name>";
   
       @Bean
       EventHubClientBuilder eventHubClientBuilder() {
           return new EventHubClientBuilder().credential(EVENT_HUB_FULLY_QUALIFIED_NAMESPACE, EVENT_HUB_NAME,
               new DefaultAzureCredentialBuilder()
                   .build());
       }
   
       @Bean
       BlobContainerClientBuilder blobContainerClientBuilder() {
           return new BlobContainerClientBuilder().credential(new DefaultAzureCredentialBuilder()
                                                      .build())
                                                  .endpoint(STORAGE_ACCOUNT_ENDPOINT)
                                                  .containerName(STORAGE_CONTAINER_NAME);
       }
   
       @Bean
       BlobContainerAsyncClient blobContainerAsyncClient(BlobContainerClientBuilder blobContainerClientBuilder) {
           return blobContainerClientBuilder.buildAsyncClient();
       }
   
       @Bean
       EventProcessorClientBuilder eventProcessorClientBuilder(BlobContainerAsyncClient blobContainerAsyncClient) {
           return new EventProcessorClientBuilder().credential(EVENT_HUB_FULLY_QUALIFIED_NAMESPACE, EVENT_HUB_NAME,
                                                       new DefaultAzureCredentialBuilder()
                                                           .build())
                                                   .consumerGroup(CONSUMER_GROUP)
                                                   .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient))
                                                   .processEvent(EventHubClientConfiguration::processEvent)
                                                   .processError(EventHubClientConfiguration::processError);
       }
   
       @Bean
       EventHubProducerClient eventHubProducerClient(EventHubClientBuilder eventHubClientBuilder) {
           return eventHubClientBuilder.buildProducerClient();
   
       }
   
       @Bean
       EventProcessorClient eventProcessorClient(EventProcessorClientBuilder eventProcessorClientBuilder) {
           return eventProcessorClientBuilder.buildEventProcessorClient();
       }
   
       public static void processEvent(EventContext eventContext) {
           LOGGER.info("Processing event from partition {} with sequence number {} with body: {}",
               eventContext.getPartitionContext().getPartitionId(), eventContext.getEventData().getSequenceNumber(),
               eventContext.getEventData().getBodyAsString());
       }
   
       public static void processError(ErrorContext errorContext) {
           LOGGER.info("Error occurred in partition processor for partition {}, {}",
               errorContext.getPartitionContext().getPartitionId(),
               errorContext.getThrowable());
       }
   
   }
   

2. Infoga EventProcessorClient och EventHubProducerClient i din Spring-applikation, som visas i följande exempel:

   import com.azure.messaging.eventhubs.EventData;
   import com.azure.messaging.eventhubs.EventHubProducerClient;
   import com.azure.messaging.eventhubs.EventProcessorClient;
   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 java.util.Collections;
   import java.util.concurrent.TimeUnit;
   
   @SpringBootApplication
   public class EventHubClientApplication implements CommandLineRunner {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubClientApplication.class);
       private final EventHubProducerClient eventHubProducerClient;
       private final EventProcessorClient eventProcessorClient;
   
       public EventHubClientApplication(EventHubProducerClient eventHubProducerClient,
                                        EventProcessorClient eventProcessorClient) {
           this.eventHubProducerClient = eventHubProducerClient;
           this.eventProcessorClient = eventProcessorClient;
       }
   
       public static void main(String[] args) {
           SpringApplication.run(EventHubClientApplication.class, args);
       }
   
       @Override
       public void run(String... args) throws Exception {
           eventProcessorClient.start();
           // Wait for the processor client to be ready
           TimeUnit.SECONDS.sleep(10);
   
           eventHubProducerClient.send(Collections.singletonList(new EventData("Hello World")));
           LOGGER.info("Successfully sent a message to Event Hubs.");
           eventHubProducerClient.close();
           LOGGER.info("Stopping and closing the processor");
           eventProcessorClient.stop();
       }
   
   }
   

3. Starta programmet. Du visas loggar som liknar följande exempel:

   Successfully sent a message to Event Hubs.
   ...
   Processing event from partition 0 with sequence number 0 with body: Hello World
   ...
   Stopping and closing the processor.
   

I följande lista visas några orsaker till varför den här koden inte är flexibel eller graciös:

• Event Hubs-namnområdet och händelsehubbens namn är hårdkodade.
• Om du använder @Value för att hämta konfigurationer från Spring-miljön kan du inte ha IDE-tips i filen application.properties .
• Om du har ett mikrotjänstscenario måste du duplicera koden i varje projekt, och det är enkelt att göra misstag och svårt att vara konsekvent.

Lyckligtvis är det inte nödvändigt att bygga klientbönorna själv med Spring Cloud Azure. I stället kan du mata in dem direkt och använda de konfigurationsegenskaper som du redan är bekant med för att konfigurera Lagringskö. Mer information finns i Spring Cloud Azure configuration.

Spring Cloud Azure tillhandahåller även följande globala konfigurationer för olika scenarier. Mer information finns i avsnittet Global configuration for Azure Service SDK i Spring Cloud Azure configuration.

• Proxyalternativ.
• Återförsöksalternativ.
• Alternativ för AMQP-transportklient.

Du kan också ansluta till olika Azure moln. Mer information finns i Anslut till olika Azure moln.

Använd Spring Messaging och Azure Event Hubs

Modulen Spring Messaging Azure Event Hubs ger stöd för ramverket Spring Messaging med Event Hubs.

Om du använder Spring Messaging-Azure Event Hubs kan du använda följande funktioner:

• EventHubsTemplate: Skicka meddelanden till Event Hubs asynkront och synkront.
• @EventHubsListener: Markera en metod som mål för en Event Hubs-meddelandelyssnare på destinationen.

Den här guiden visar hur du använder Spring Messaging Azure Event Hubs för att skicka meddelanden till och ta emot meddelanden från Event Hubs.

Lägga till beroenden

Om du vill installera Spring Messaging Azure Event Hubs-modulen lägger du till följande beroenden i filen pom.xml:

• Spring Cloud Azure Materialförteckning (BOM):

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

  Anteckning

  Om du använder Spring Boot 4.0.x anger du spring-cloud-azure-dependencies versionen till 7.4.0.

  Om du använder Spring Boot 3.5.x anger du spring-cloud-azure-dependencies versionen till 6.5.0.

  Om du använder Spring Boot 3.1.x-3.5.x måste du ange spring-cloud-azure-dependencies versionen till 5.25.0.

  Om du använder Spring Boot 2.x måste du ange spring-cloud-azure-dependencies versionen till 4.20.0.

  Den här materialspecifikationen (BOM) bör konfigureras i avsnittet <dependencyManagement> i din pom.xml-fil. Detta säkerställer att alla Spring Cloud-Azure beroenden använder samma version.

  Mer information om vilken version som används för den här BOM finns genom att läsa Which Version of Spring Cloud Azure Should I Use.

• Spring Cloud Azure-startpaket, Spring Messaging Event Hubs och Azure Event Hubs Checkpoint Store-komponenter:

  <dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter</artifactId>
  </dependency>
  <dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-messaging-azure-eventhubs</artifactId>
  </dependency>
  <dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
  </dependency>
  

Koda programmet för att skicka och ta emot meddelanden

Om du vill skicka meddelanden till och ta emot meddelanden från Event Hubs konfigurerar du programmet med hjälp av följande steg:

1. Använd följande egenskapsinställningar för att konfigurera Event Hubs-namnområdet och lagringsbloben:

   spring.cloud.azure.eventhubs.namespace=<your event-hubs-namespace>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=<your-storage-account-name>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=<your-storage-account-container-name>
   

2. Skapa en ny klass ConsumerService Java enligt följande exempel. Den här klassen används för att definiera en meddelandemottagare. Se till att ersätta platshållaren <your-event-hub-name> med ditt faktiska värde.

   import com.azure.spring.messaging.eventhubs.implementation.core.annotation.EventHubsListener;
   import org.springframework.stereotype.Service;
   
   @Service
   public class ConsumerService {
   
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
       private static final String CONSUMER_GROUP = "$DEFAULT";
   
       @EventHubsListener(destination = EVENT_HUB_NAME, group = CONSUMER_GROUP)
       public void handleMessageFromEventHub(String message) {
           System.out.printf("New message received: %s%n", message);
       }
   
   }
   

3. Koppla en avsändare och en mottagare för att skicka och ta emot meddelanden med Spring, som du ser i följande exempel. Se till att ersätta platshållaren <your-event-hub-name> med ditt faktiska värde.

   import com.azure.spring.messaging.eventhubs.core.EventHubsTemplate;
   import com.azure.spring.messaging.implementation.annotation.EnableAzureMessaging;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.boot.SpringApplication;
   import org.springframework.boot.autoconfigure.SpringBootApplication;
   import org.springframework.context.ConfigurableApplicationContext;
   import org.springframework.messaging.support.MessageBuilder;
   
   @SpringBootApplication
   @EnableAzureMessaging
   public class EventHubMessagingApplication {
   
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubMessagingApplication.class);
   
       public static void main(String[] args) {
           ConfigurableApplicationContext applicationContext = SpringApplication.run(EventHubMessagingApplication.class);
           EventHubsTemplate eventHubsTemplate = applicationContext.getBean(EventHubsTemplate.class);
           LOGGER.info("Sending a message to the Event Hubs.");
           eventHubsTemplate.sendAsync(EVENT_HUB_NAME, MessageBuilder.withPayload("Hello world").build()).subscribe();
       }
   
   }
   

   Tips

   Kom ihåg att lägga till annoteringen @EnableAzureMessaging , som utlöser identifieringen av metoder som kommenterats med @EventHubsListener, och som skapar containern för meddelandelyssnare i bakgrunden.

4. Starta programmet. Du visas loggar som liknar följande exempel:

   Sending a message to the Event Hubs.
   New message received: Hello world
   

Använda Spring Integration Azure Event Hubs

Modulen Spring Integration Azure Event Hubs ger stöd för ramverket Spring Integration med Event Hubs.

Om spring-programmet använder Spring Integration-meddelandekanaler kan du dirigera meddelanden mellan dina meddelandekanaler och Event Hubs med hjälp av kanalkort.

En inkommande kanaladapter vidarebefordrar meddelanden från en händelsehubb till en meddelandekanal. En utgående kanaladapter publicerar meddelanden från en kanal till en händelsehubb.

Den här guiden visar hur du använder Spring Integration Azure Event Hubs för att skicka meddelanden till och ta emot meddelanden från Event Hubs.

Lägga till beroenden

Om du vill installera Modulen Spring Cloud Azure Event Hubs Integration Starter lägger du till följande beroenden i filen pom.xml:

• Spring Cloud Azure Materialförteckning (BOM):

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

  Anteckning

  Om du använder Spring Boot 4.0.x anger du spring-cloud-azure-dependencies versionen till 7.4.0.

  Om du använder Spring Boot 3.5.x anger du spring-cloud-azure-dependencies versionen till 6.5.0.

  Om du använder Spring Boot 3.1.x-3.5.x måste du ange spring-cloud-azure-dependencies versionen till 5.25.0.

  Om du använder Spring Boot 2.x måste du ange spring-cloud-azure-dependencies versionen till 4.20.0.

  Den här materialspecifikationen (BOM) bör konfigureras i avsnittet <dependencyManagement> i din pom.xml-fil. Detta säkerställer att alla Spring Cloud-Azure beroenden använder samma version.

  Mer information om vilken version som används för den här BOM finns genom att läsa Which Version of Spring Cloud Azure Should I Use.

• Spring Cloud Azure Event Hubs Integration-artefakten:

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

Koda programmet för att skicka och ta emot meddelanden

Om du vill skicka meddelanden till och ta emot meddelanden från Event Hubs konfigurerar du programmet med hjälp av följande steg:

1. Använd följande egenskapsinställningar för att konfigurera Event Hubs-namnområdet och lagringsbloben:

   spring.cloud.azure.eventhubs.namespace=<your event-hubs-namespace>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=<your-storage-account-name>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=<your-storage-account-container-name>
   

2. Skapa en ny klass MessageReceiveConfiguration Java enligt följande exempel. Den här klassen används för att definiera en meddelandemottagare. Se till att ersätta platshållaren <your-event-hub-name> med ditt faktiska värde.

   import com.azure.spring.integration.eventhubs.inbound.EventHubsInboundChannelAdapter;
   import com.azure.spring.messaging.eventhubs.core.EventHubsProcessorFactory;
   import com.azure.spring.messaging.eventhubs.core.checkpoint.CheckpointConfig;
   import com.azure.spring.messaging.eventhubs.core.checkpoint.CheckpointMode;
   import com.azure.spring.messaging.eventhubs.core.listener.EventHubsMessageListenerContainer;
   import com.azure.spring.messaging.eventhubs.core.properties.EventHubsContainerProperties;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.beans.factory.annotation.Qualifier;
   import org.springframework.context.annotation.Bean;
   import org.springframework.context.annotation.Configuration;
   import org.springframework.integration.annotation.ServiceActivator;
   import org.springframework.integration.channel.DirectChannel;
   import org.springframework.messaging.MessageChannel;
   
   @Configuration
   public class MessageReceiveConfiguration {
   
       private static final String INPUT_CHANNEL = "input";
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
       private static final String CONSUMER_GROUP = "$Default";
       private static final Logger LOGGER = LoggerFactory.getLogger(MessageReceiveConfiguration.class);
   
       @ServiceActivator(inputChannel = INPUT_CHANNEL)
       public void messageReceiver(byte[] payload) {
           String message = new String(payload);
           LOGGER.info("New message received: {}", message);
       }
   
       @Bean
       public EventHubsMessageListenerContainer messageListenerContainer(EventHubsProcessorFactory processorFactory) {
           EventHubsContainerProperties containerProperties = new EventHubsContainerProperties();
           containerProperties.setEventHubName(EVENT_HUB_NAME);
           containerProperties.setConsumerGroup(CONSUMER_GROUP);
           containerProperties.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));
           return new EventHubsMessageListenerContainer(processorFactory, containerProperties);
       }
   
       @Bean
       public EventHubsInboundChannelAdapter messageChannelAdapter(@Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
                                                                   EventHubsMessageListenerContainer listenerContainer) {
           EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(listenerContainer);
           adapter.setOutputChannel(inputChannel);
           return adapter;
       }
   
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   
   }
   

3. Skapa en ny klass MessageSendConfiguration Java enligt följande exempel. Den här klassen används för att definiera en meddelandesändare. Se till att ersätta platshållaren <your-event-hub-name> med ditt faktiska värde.

   import com.azure.spring.integration.core.handler.DefaultMessageHandler;
   import com.azure.spring.messaging.eventhubs.core.EventHubsTemplate;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.context.annotation.Bean;
   import org.springframework.context.annotation.Configuration;
   import org.springframework.integration.annotation.MessagingGateway;
   import org.springframework.integration.annotation.ServiceActivator;
   import org.springframework.messaging.MessageHandler;
   import org.springframework.util.concurrent.ListenableFutureCallback;
   
   @Configuration
   public class MessageSendConfiguration {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(MessageSendConfiguration.class);
       private static final String OUTPUT_CHANNEL = "output";
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
   
       @Bean
       @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
       public MessageHandler messageSender(EventHubsTemplate eventHubsTemplate) {
           DefaultMessageHandler handler = new DefaultMessageHandler(EVENT_HUB_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;
       }
   
       @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
       public interface EventHubOutboundGateway {
           void send(String text);
       }
   
   }
   

4. Koppla en avsändare och en mottagare för att skicka och ta emot meddelanden med Spring, som du ser i följande exempel:

   import org.springframework.boot.SpringApplication;
   import org.springframework.boot.autoconfigure.SpringBootApplication;
   import org.springframework.context.ConfigurableApplicationContext;
   import org.springframework.context.annotation.Configuration;
   import org.springframework.integration.config.EnableIntegration;
   
   @SpringBootApplication
   @EnableIntegration
   @Configuration(proxyBeanMethods = false)
   public class EventHubIntegrationApplication {
   
       public static void main(String[] args) {
           ConfigurableApplicationContext applicationContext = SpringApplication.run(EventHubIntegrationApplication.class, args);
           MessageSendConfiguration.EventHubOutboundGateway outboundGateway = applicationContext.getBean(MessageSendConfiguration.EventHubOutboundGateway.class);
           outboundGateway.send("Hello World");
       }
   }
   

   Tips

   Kom ihåg att lägga till anteckningen @EnableIntegration , som möjliggör Spring Integration-infrastrukturen.

5. Starta programmet. Du visas loggar som liknar följande exempel:

   Message was sent successfully.
   New message received: Hello World
   

Använda Spring Cloud Azure Stream Event Hubs Binder

Om du vill anropa Event Hubs-API:et i ett Spring Cloud Stream-program använder du modulen Spring Cloud Azure Event Hubs Stream Binder.

Den här guiden visar hur du använder Spring Cloud Stream Event Hubs Binder för att skicka meddelanden till och ta emot meddelanden från Event Hubs.

Lägga till beroenden

Om du vill installera Spring Cloud Azure Event Hubs Stream Binder-modulen lägger du till följande beroenden i filen pom.xml:

• Spring Cloud Azure Materialförteckning (BOM):

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

  Anteckning

  Om du använder Spring Boot 4.0.x anger du spring-cloud-azure-dependencies versionen till 7.4.0.

  Om du använder Spring Boot 3.5.x anger du spring-cloud-azure-dependencies versionen till 6.5.0.

  Om du använder Spring Boot 3.1.x-3.5.x måste du ange spring-cloud-azure-dependencies versionen till 5.25.0.

  Om du använder Spring Boot 2.x måste du ange spring-cloud-azure-dependencies versionen till 4.20.0.

  Den här materialspecifikationen (BOM) bör konfigureras i avsnittet <dependencyManagement> i din pom.xml-fil. Detta säkerställer att alla Spring Cloud-Azure beroenden använder samma version.

  Mer information om vilken version som används för den här BOM finns genom att läsa Which Version of Spring Cloud Azure Should I Use.

• Spring Cloud Azure Event Hubs Stream Binder-artefaktet:

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

Koda programmet för att skicka och ta emot meddelanden

Om du vill skicka meddelanden till och ta emot meddelanden från Event Hubs konfigurerar du programmet med hjälp av följande steg:

1. Använd följande egenskapsinställningar för att konfigurera Event Hubs-namnområdet och lagringsbloben:

   spring.cloud.azure.eventhubs.namespace=<your event-hubs-namespace>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=<your-storage-account-name>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=<your-storage-account-container-name>
   

2. Skapa meddelandemottagaren.

   Om du vill använda ditt program som händelsemottagare konfigurerar du indatabindningen genom att utföra följande uppgifter:

   • Deklarera en Consumer böna som definierar logik för meddelandehantering. Följande böna heter Consumertill exempel consume :

     @Bean
     public Consumer<Message<String>> consume() {
          return message -> {
              System.out.printf("New message received: %s%n", message.getPayload());
          };
     }
     

   • Lägg till följande konfiguration för att ange Event Hub namnet för konsumtion. Se till att ersätta platshållaren <your-event-hub-name> med ditt faktiska värde.

     # name for the above `Consumer` bean
     spring.cloud.stream.function.definition=consume
     spring.cloud.stream.bindings.consume-in-0.destination=<your-event-hub-name>
     spring.cloud.stream.bindings.consume-in-0.group=$Default
     spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode=MANUAL
     

3. Skapa meddelandesändaren.

   Om du vill använda ditt program som en händelsekälla konfigurerar du utdatabindningen genom att utföra följande uppgifter:

   • Definiera en Supplier böna som definierar var meddelanden kommer från ditt program, enligt följande exempel:

     @Bean
     public Supplier<Message<String>> supply() {
         return () -> {
             System.out.println("Sending a message.");
             return MessageBuilder.withPayload("Hello world").build();
         };
     }
     

   • Lägg till följande konfiguration för att ange namnet på sändningen Event Hub . Se till att ersätta platshållaren <your-event-hub-name> med ditt faktiska värde.

     # "consume" is added from the above step
     spring.cloud.stream.function.definition=consume;supply
     spring.cloud.stream.bindings.supply-out-0.destination=<your-event-hub-name>
     

4. Starta programmet. Du visas loggar som liknar följande exempel:

   Sending a message.
   New message received: Hello world.
   

Använda Spring Kafka med Azure Event Hubs

Event Hubs tillhandahåller en Kafka-slutpunkt som dina befintliga Kafka-baserade program kan använda. Den här metoden är ett alternativ till att köra ett eget Kafka-kluster. Event Hubs fungerar med många av dina befintliga Kafka-program. Mer information finns i Event Hubs för Apache Kafka.

Den här guiden visar hur du använder Azure Event Hubs och Spring Kafka för att skicka meddelanden till och ta emot meddelanden från Event Hubs.

Lägga till beroenden

Om du vill installera Spring Cloud-modulerna Azure starter och Spring Kafka lägger du till följande beroenden i filen pom.xml:

• Spring Cloud Azure Materialförteckning (BOM):

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

  Anteckning

  Om du använder Spring Boot 4.0.x anger du spring-cloud-azure-dependencies versionen till 7.4.0.

  Om du använder Spring Boot 3.5.x anger du spring-cloud-azure-dependencies versionen till 6.5.0.

  Om du använder Spring Boot 3.1.x-3.5.x måste du ange spring-cloud-azure-dependencies versionen till 5.25.0.

  Om du använder Spring Boot 2.x måste du ange spring-cloud-azure-dependencies versionen till 4.20.0.

  Den här materialspecifikationen (BOM) bör konfigureras i avsnittet <dependencyManagement> i din pom.xml-fil. Detta säkerställer att alla Spring Cloud-Azure beroenden använder samma version.

  Mer information om vilken version som används för den här BOM finns genom att läsa Which Version of Spring Cloud Azure Should I Use.

• Spring Cloud-Azure starter och Spring Kafka-artefakt:

  <dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
  </dependency>
  

Koda programmet för att skicka och ta emot meddelanden

Om du vill skicka meddelanden till och ta emot meddelanden från Event Hubs konfigurerar du programmet med hjälp av följande steg:

1. Använd följande egenskapsinställning för att konfigurera Event Hubs-namnområdet:

   spring.kafka.bootstrap-servers=<your event-hubs-namespace>.servicebus.windows.net:9093
   

2. Använd KafkaTemplate för att skicka meddelanden och @KafkaListener ta emot meddelanden, som du ser i följande exempel. Se till att ersätta platshållaren <your-event-hub-name> med ditt faktiska värde.

   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.kafka.annotation.KafkaListener;
   import org.springframework.kafka.core.KafkaTemplate;
   
   @SpringBootApplication
   public class EventHubKafkaApplication implements CommandLineRunner {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubKafkaApplication.class);
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
       private static final String CONSUMER_GROUP = "$Default";
       private final KafkaTemplate<String, String> kafkaTemplate;
   
       public EventHubKafkaApplication(KafkaTemplate<String, String> kafkaTemplate) {
           this.kafkaTemplate = kafkaTemplate;
       }
   
       public static void main(String[] args) {
           SpringApplication.run(EventHubKafkaApplication.class, args);
       }
   
       @Override
       public void run(String... args) {
           kafkaTemplate.send(EVENT_HUB_NAME, "Hello World");
           LOGGER.info("Message was sent successfully.");
       }
   
       @KafkaListener(topics = EVENT_HUB_NAME, groupId = CONSUMER_GROUP)
       public void receive(String message) {
           LOGGER.info("New message received: {}", message);
       }
   
   }
   

3. Starta programmet. Du visas loggar som liknar följande exempel:

   Message was sent successfully.
   New message received: Hello world
   

Använda Spring Cloud Stream Kafka Binder med Azure Event Hubs

Spring Cloud Stream är ett ramverk som gör det möjligt för programutvecklare att skriva meddelandedrivna mikrotjänster. Bryggan mellan ett meddelandesystem och Spring Cloud Stream sker via bindningsabstraktionen. Binders finns för flera meddelandesystem, men en av de vanligaste pärmarna är för Apache Kafka.

Den här guiden visar hur du använder Azure Event Hubs och Spring Cloud Stream Kafka Binder för att skicka meddelanden till och ta emot meddelanden från Event Hubs.

Lägga till beroenden

Om du vill installera Spring Cloud-modulerna Azure starter och Spring Cloud Stream binder Kafka lägger du till följande beroenden i filen pom.xml:

• Spring Cloud Azure Materialförteckning (BOM):

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

  Anteckning

  Om du använder Spring Boot 4.0.x anger du spring-cloud-azure-dependencies versionen till 7.4.0.

  Om du använder Spring Boot 3.5.x anger du spring-cloud-azure-dependencies versionen till 6.5.0.

  Om du använder Spring Boot 3.1.x-3.5.x måste du ange spring-cloud-azure-dependencies versionen till 5.25.0.

  Om du använder Spring Boot 2.x måste du ange spring-cloud-azure-dependencies versionen till 4.20.0.

  Den här materialspecifikationen (BOM) bör konfigureras i avsnittet <dependencyManagement> i din pom.xml-fil. Detta säkerställer att alla Spring Cloud-Azure beroenden använder samma version.

  Mer information om vilken version som används för den här BOM finns genom att läsa Which Version of Spring Cloud Azure Should I Use.

• Spring Cloud Azure startartefakt:

  <dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
  </dependency>
  

Koda programmet för att skicka och ta emot meddelanden

Om du vill skicka meddelanden till och ta emot meddelanden från Event Hubs konfigurerar du programmet med hjälp av följande steg:

1. Använd följande egenskapsinställning för att konfigurera Kafka-koordinatorn:

   spring.cloud.stream.kafka.binder.brokers=<your event-hubs-namespace>.servicebus.windows.net:9093
   

2. Skapa meddelandemottagaren.

   Om du vill använda ditt program som händelsemottagare konfigurerar du indatabindningen genom att utföra följande uppgifter:

   • Deklarera en Consumer böna som definierar logik för meddelandehantering. Följande böna heter Consumertill exempel consume :

     @Bean
     public Consumer<Message<String>> consume() {
         return message -> {
             System.out.printf("New message received: %s%n", message.getPayload());
         };
     }
     

   • Lägg till följande konfiguration för att ange Event Hub namnet för konsumtion. Se till att ersätta platshållaren <your-event-hub-name> med ditt faktiska värde.

     # name for the above `Consumer` bean
     spring.cloud.stream.function.definition=consume
     spring.cloud.stream.bindings.consume-in-0.destination=<your-event-hub-name>
     spring.cloud.stream.bindings.consume-in-0.group=$Default
     

3. Skapa meddelandesändaren.

   Om du vill använda ditt program som en händelsekälla konfigurerar du utdatabindningen genom att utföra följande uppgifter:

   • Definiera en Supplier böna som definierar var meddelanden kommer från ditt program, enligt följande exempel:

     @Bean
     public Supplier<Message<String>> supply() {
         return () -> {
             System.out.println("Sending a message.");
             return MessageBuilder.withPayload("Hello world").build();
         };
     }
     

   • Lägg till följande konfiguration för att ange namnet på sändningen Event Hub . Se till att ersätta platshållaren <your-event-hub-name> med ditt faktiska värde.

     # "consume" is added from the above step
     spring.cloud.stream.function.definition=consume;supply
     spring.cloud.stream.bindings.supply-out-0.destination=<your-event-hub-name>
     

4. Starta programmet. Du visas loggar som liknar följande exempel:

   Sending a message.
   New message received: Hello world.
   

Distribuera till Azure Spring Apps

Nu när Spring Boot-programmet körs lokalt är det dags att flytta det till produktion. Azure Spring Apps gör det enkelt att distribuera Spring Boot-program till Azure utan några kodändringar. Tjänsten hanterar infrastrukturen för Spring-program så att utvecklare kan fokusera på sin kod. Azure Spring Apps tillhandahåller livscykelhantering med omfattande övervakning och diagnostik, konfigurationshantering, tjänstidentifiering, CI/CD-integrering, blågröna distributioner med mera. Information om hur du distribuerar ditt program till Azure Spring Apps finns i Distribuera ditt första program till Azure Spring Apps.

Nästa steg

Azure för Spring-utvecklareSpring Cloud Azure Event Hubs Samples

Se även

Mer information om de andra Spring Boot-startarna som är tillgängliga för Microsoft Azure finns i Vad är Spring Cloud Azure?