Použití služby Event Hubs v aplikacích Spring

V tomto článku se dozvíte, jak používat Azure Event Hubs v aplikacích Java vytvořených pomocí Spring Framework.

Azure Event Hubs je platforma pro streamování velkých objemů dat a služba pro příjem událostí. Dokáže přijímat a zpracovávat miliony událostí za sekundu. Data odeslaná do centra událostí je možné transformovat a uložit pomocí libovolného poskytovatele analýz v reálném čase nebo adaptérů pro dávkové zpracování a ukládání.

Spring Cloud Azure poskytuje různé moduly pro odesílání zpráv do a přijímání zpráv ze služby Event Hubs pomocí architektur Spring.

Následující moduly můžete použít nezávisle nebo je kombinovat pro různé případy použití:

• Spring Cloud Azure Event Hubs Starter umožňuje odesílat a přijímat zprávy pomocí klientské knihovny služby Event Hubs Java SDK s funkcemi Spring Boot.

• Spring Messaging Azure Event Hubs umožňuje pracovat se službou Event Hubs prostřednictvím rozhraní API Spring Messaging API.

• Spring Integration Azure Event Hubs umožňuje připojit zprávové kanály Spring Integration k Event Hubs.

• Spring Cloud Azure Stream Event Hubs Binder umožňuje používat službu Event Hubs jako middleware pro zasílání zpráv v aplikacích Spring Cloud Stream.

• Spring Kafka s Azure Event Hubs umožňuje používat Spring Kafka k odesílání a přijímání zpráv ze služby Event Hubs.

• Spring Cloud Stream Kafka Binder s Azure Event Hubs umožňuje odesílat a přijímat zprávy přes Spring Cloud Stream Kafka Binder se službou Event Hubs.

Požadavky

• Předplatné Azure – vytvořte si ho zdarma.

• Java Development Kit (JDK) verze 8 nebo vyšší.

• Apache Maven

• Azure CLI.

• Instance služby Azure Event Hubs. Další informace najdete v tématu Quickstart: Vytvoření centra událostí pomocí portálu Azure.

• Účet Azure Storage pro kontrolní body služby Event Hubs. Další informace najdete v článku o vytvoření účtu úložiště.

• Aplikace Spring Boot. Pokud ho nemáte, vytvořte projekt Maven pomocí aplikace Spring Initializr. Nezapomeňte vybrat Maven Project a v části Dependencies přidejte závislost Spring Web a pak vyberte Java verze 8 nebo vyšší.

Poznámka:

Pokud chcete účtu udělit přístup k prostředkům, přiřaďte v Azure Event Hubs roli Azure Event Hubs Data Receiver a Azure Event Hubs Data Sender k účtu Microsoft Entra, který právě používáte. Potom v účtu Azure Storage přiřaďte roli Storage Blob Data Contributor k účtu Microsoft Entra, který právě používáte. Další informace o udělení přístupových rolí najdete v tématu Přiřazení rolí Azure pomocí portálu Azure a Autorizace přístupu k prostředkům služby Event Hubs pomocí Microsoft Entra ID.

Důležité

K dokončení kroků v tomto kurzu se vyžaduje Spring Boot verze 2.5 nebo vyšší.

Příprava místního prostředí

V tomto kurzu konfigurace a kód nemají žádné ověřovací operace. Připojení ke službě Azure ale vyžaduje ověření. K dokončení ověřování je potřeba použít klientskou knihovnu Azure Identity. Spring Cloud Azure používá DefaultAzureCredential, které knihovna identit Azure poskytuje, aby vám pomohla získat přihlašovací údaje bez jakýchkoli změn kódu.

DefaultAzureCredential podporuje více metod ověřování a určuje, kterou metodu použít za běhu. Tento přístup umožňuje vaší aplikaci používat různé metody ověřování v různých prostředích , jako jsou místní nebo produkční prostředí, bez implementace kódu specifického pro prostředí. Další informace najdete v části DefaultAzureCredential v části Authenticate Azure hostované Java aplikace.

Pokud chcete k dokončení ověřování v místních vývojových prostředích použít Azure CLI, IntelliJ nebo jiné metody, přečtěte si Azure ověřování v Java vývojových prostředích. K dokončení ověřování v Azure hostitelských prostředích doporučujeme použít spravovanou identitu. Další informace najdete v tématu Co jsou spravované identity pro prostředky Azure?

Použijte startér Spring Cloud Azure Event Hubs

Modul Spring Cloud Azure Event Hubs Starter importuje klientskou knihovnu Event Hubs Java s frameworkem Spring Boot. Můžete používat Spring Cloud Azure a Azure SDK společně ve vzoru, který se nevylučuje. Proto můžete ve své aplikaci Spring dál používat rozhraní API klienta služby Event Hubs Java.

Přidejte závislosti

Pokud chcete nainstalovat modul Spring Cloud Azure Event Hubs Starter, přidejte do souboru pom.xml0 následující závislosti:

• Spring Cloud Azure 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>
  

  Poznámka:

  Pokud používáte Spring Boot 4.0.x, nastavte spring-cloud-azure-dependencies verzi na 7.4.0hodnotu .

  Pokud používáte Spring Boot 3.5.x, nastavte spring-cloud-azure-dependencies verzi na 6.5.0hodnotu .

  Pokud používáte Spring Boot 3.1.x-3.5.x, nezapomeňte nastavit verzi na spring-cloud-azure-dependencies.

  Pokud používáte Spring Boot 2.x, nezapomeňte nastavit spring-cloud-azure-dependencies verzi na 4.20.0.

  Tato faktura materiálu (BOM) by měla být nakonfigurována v <dependencyManagement> části vašeho pom.xml souboru. Tím zajistíte, že všechny závislosti Spring Cloud Azure budou používat stejnou verzi.

  Další informace o verzi použité pro tento kusovník najdete v tématu Kterou verzi Spring Cloud Azure bych měl použít.

• Artefakt Spring Cloud Azure Event Hubs

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

Kódování aplikace pro odesílání a příjem zpráv

V této příručce se naučíte používat službu Event Hubs Java klienty v kontextu aplikace Spring. Tady představujeme následující dvě možnosti:

• Použijte automatickou konfiguraci Spring Boot a využijte předem připravené klienty ze Spring kontextu (doporučeno).
• Programově sestavte klienta.

Způsob automatického propojování klientských beanů z kontejneru Spring IoC má následující výhody, které vám mohou poskytnout flexibilnější a efektivnější zkušenost při vývoji aplikací s klienty Event Hubs:

• Použije externalizovanou konfiguraci, abyste mohli pracovat se stejným kódem aplikace v různých prostředích.
• Na architekturu Spring Boot můžete delegovat proces učení modelu tvůrce a registraci tohoto klienta do kontextu aplikace. Toto delegování vám umožní zaměřit se na to, jak přizpůsobit použití klientů vašim obchodním požadavkům.
• Indikátor stavu můžete snadno použít ke kontrole stavu a stavu aplikace a interních komponent.

Následující části obsahují příklady kódu, které vám ukážou, jak se dají použít EventProcessorClient , a EventHubProducerClient se dvěma alternativami.

Poznámka:

Azure Java SDK pro službu Event Hubs poskytuje více klientům pro interakci se službou Event Hubs. Úvodní sada také poskytuje automatickou konfiguraci pro všechny klienty služby Event Hubs i pro tvůrce klientů. Tento článek používá pouze EventProcessorClient a EventHubProducerClient jako příklady.

Použití funkce Spring Boot Autoconfiguration

Pokud chcete odesílat zprávy do služby Event Hubs a přijímat je, nakonfigurujte aplikaci pomocí následujícího postupu:

1. K nastavení oboru názvů Event Hubs a názvu centra událostí použijte následující parametry:

   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. Vytvořte novou třídu EventHubProcessorClientConfiguration Java, jak je znázorněno v následujícím příkladu. Tato třída slouží k registraci obslužné rutiny zpráv a chyb pro 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. Vložte EventProcessorClient a EventHubProducerClient do své aplikace Spring a volejte související rozhraní API pro odesílání a přijímání zpráv, jak je znázorněno v následujícím příkladu:

   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. Spusťte aplikaci. Zobrazí se protokoly podobné následujícímu příkladu:

   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.
   

Programové sestavení klienta

Klientské komponenty můžete vytvořit sami, ale proces je komplikovaný. V aplikacích Spring Boot musíte spravovat vlastnosti, zjistit vzor tvůrce a zaregistrovat klienta do kontextu aplikace Spring. Následující kroky ukazují, jak to udělat:

1. Vytvořte novou třídu EventHubClientConfiguration Java, jak je znázorněno v následujícím příkladu. Tato třída se používá k deklaraci EventProcessorClient a EventHubProducerClient beanů. Nezapomeňte nahradit zástupné symboly <your event-hubs-namespace>, <your-event-hub-name>, <your-storage-account-name> a <your-storage-account-container-name> skutečnými hodnotami.

   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. EventProcessorClient a EventHubProducerClient vložte do vaší Spring aplikace, jak je znázorněno v následujícím příkladu:

   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. Spusťte aplikaci. Zobrazí se protokoly podobné následujícímu příkladu:

   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.
   

Následující seznam uvádí některé důvody, proč tento kód není flexibilní nebo elegantní:

• Obor názvů služby Event Hubs a název centra událostí jsou pevně zakódované.
• Pokud používáte @Value k získání konfigurací z prostředí Spring, nemůžete mít v souboru application.properties nápovědy IDE.
• Pokud máte scénář mikroslužby, musíte kód v každém projektu duplikovat a je snadné udělat chyby a těžko být konzistentní.

Naštěstí není nezbytné vytvářet klientské komponenty sami pomocí služby Azure Spring Cloud. Místo toho je můžete přímo vložit a použít konfigurační vlastnosti, které již znáte, pro konfiguraci fronty úložiště. Další informace najdete v tématu konfigurace Spring Cloud Azure.

Spring Cloud Azure také poskytuje následující globální konfigurace pro různé scénáře. Další informace najdete v části Globální konfigurace sad SDK služby Azure v konfiguraci Spring Cloud Azure.

• Možnosti proxy serveru.
• Možnosti opakování
• Možnosti přenosového klienta AMQP

Můžete se také připojit k různým Azure cloudům. Další informace najdete v tématu Pojení k různým cloudům Azure.

Použijte Spring Messaging s Azure Event Hubs

Modul Spring Messaging Azure Event Hubs poskytuje podporu pro Spring Messaging framework se službou Event Hubs.

Pokud používáte spring messaging Azure Event Hubs, můžete použít následující funkce:

• EventHubsTemplate: Posílejte zprávy do služby Event Hubs asynchronně a synchronně.
• @EventHubsListener: Označte metodu jako cíl pro posluchače zpráv v Event Hubs na cílové stanici.

V této příručce se dozvíte, jak používat spring messaging Azure Event Hubs k odesílání a přijímání zpráv ze služby Event Hubs.

Přidejte závislosti

Pokud chcete nainstalovat modul Azure Event Hubs Spring Messaging, přidejte do souboru pom.xml0 následující závislosti:

• Spring Cloud Azure 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>
  

  Poznámka:

  Pokud používáte Spring Boot 4.0.x, nastavte spring-cloud-azure-dependencies verzi na 7.4.0hodnotu .

  Pokud používáte Spring Boot 3.5.x, nastavte spring-cloud-azure-dependencies verzi na 6.5.0hodnotu .

  Pokud používáte Spring Boot 3.1.x-3.5.x, nezapomeňte nastavit verzi na spring-cloud-azure-dependencies.

  Pokud používáte Spring Boot 2.x, nezapomeňte nastavit spring-cloud-azure-dependencies verzi na 4.20.0.

  Tato faktura materiálu (BOM) by měla být nakonfigurována v <dependencyManagement> části vašeho pom.xml souboru. Tím zajistíte, že všechny závislosti Spring Cloud Azure budou používat stejnou verzi.

  Další informace o verzi použité pro tento kusovník najdete v tématu Kterou verzi Spring Cloud Azure bych měl použít.

• Úvodní Azure Spring Cloud, služba Spring Messaging Event Hubs a artefakty úložiště kontrolních bodů Azure Event Hubs:

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

Kódování aplikace pro odesílání a příjem zpráv

Pokud chcete odesílat zprávy do služby Event Hubs a přijímat je, nakonfigurujte aplikaci pomocí následujícího postupu:

1. Ke konfiguraci namespace Event Hubs a Storage Blob použijte následující nastavení vlastností:

   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. Vytvořte novou třídu ConsumerService Java, jak je znázorněno v následujícím příkladu. Tato třída slouží k definování příjemce zprávy. Nezapomeňte zástupný symbol nahradit skutečnou <your-event-hub-name> hodnotou.

   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. Pro připojení odesílatele a příjemce k odesílání a přijímání zpráv pomocí Springu, jak je znázorněno v následujícím příkladu. Nezapomeňte zástupný symbol nahradit skutečnou <your-event-hub-name> hodnotou.

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

   Návod

   Nezapomeňte přidat poznámku @EnableAzureMessaging, která aktivuje zjišťování metod anotovaných pomocí @EventHubsListener, čímž se vytvoří kontejner pro naslouchání zpráv na pozadí.

4. Spusťte aplikaci. Zobrazí se protokoly podobné následujícímu příkladu:

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

Použití integrace Spring pro Azure Event Hubs

Modul integrace Spring Azure Event Hubs poskytuje podporu pro architekturu integrace Spring se službou Event Hubs.

Pokud vaše aplikace Spring používá kanály zpráv Spring Integration, můžete směrovat zprávy mezi kanály zpráv a službou Event Hubs pomocí adaptérů kanálů.

Adaptér příchozího kanálu předává zprávy z centra událostí do kanálu zpráv. Adaptér odchozího kanálu publikuje zprávy z kanálu zpráv do centra událostí.

V této příručce se dozvíte, jak používat spring integration Azure Event Hubs k odesílání a přijímání zpráv ze služby Event Hubs.

Přidejte závislosti

Pokud chcete nainstalovat modul Spring Cloud Azure Event Hubs Integration Starter, přidejte do souboru pom.xml0 následující závislosti:

• Spring Cloud Azure 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>
  

  Poznámka:

  Pokud používáte Spring Boot 4.0.x, nastavte spring-cloud-azure-dependencies verzi na 7.4.0hodnotu .

  Pokud používáte Spring Boot 3.5.x, nastavte spring-cloud-azure-dependencies verzi na 6.5.0hodnotu .

  Pokud používáte Spring Boot 3.1.x-3.5.x, nezapomeňte nastavit verzi na spring-cloud-azure-dependencies.

  Pokud používáte Spring Boot 2.x, nezapomeňte nastavit spring-cloud-azure-dependencies verzi na 4.20.0.

  Tato faktura materiálu (BOM) by měla být nakonfigurována v <dependencyManagement> části vašeho pom.xml souboru. Tím zajistíte, že všechny závislosti Spring Cloud Azure budou používat stejnou verzi.

  Další informace o verzi použité pro tento kusovník najdete v tématu Kterou verzi Spring Cloud Azure bych měl použít.

• Artefakt integrace Azure Event Hubs Spring Cloudu:

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

Kódování aplikace pro odesílání a příjem zpráv

Pokud chcete odesílat zprávy do služby Event Hubs a přijímat je, nakonfigurujte aplikaci pomocí následujícího postupu:

1. Ke konfiguraci namespace Event Hubs a Storage Blob použijte následující nastavení vlastností:

   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. Vytvořte novou třídu MessageReceiveConfiguration Java, jak je znázorněno v následujícím příkladu. Tato třída slouží k definování příjemce zprávy. Nezapomeňte zástupný symbol nahradit skutečnou <your-event-hub-name> hodnotou.

   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. Vytvořte novou třídu MessageSendConfiguration Java, jak je znázorněno v následujícím příkladu. Tato třída slouží k definování odesílatele zprávy. Nezapomeňte zástupný symbol nahradit skutečnou <your-event-hub-name> hodnotou.

   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. Vytvořte odesílatele a příjemce, aby odesílali a přijímali zprávy pomocí Springu, jak je znázorněno v následujícím příkladu:

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

   Návod

   Nezapomeňte přidat poznámku @EnableIntegration , která umožňuje infrastrukturu integrace Spring.

5. Spusťte aplikaci. Zobrazí se protokoly podobné následujícímu příkladu:

   Message was sent successfully.
   New message received: Hello World
   

Použití Spring Cloud Azure Stream Event Hubs Binderu

K volání rozhraní API služby Event Hubs v aplikaci Spring Cloud Stream použijte modul Spring Cloud Azure Event Hubs Stream Binder.

V této příručce se dozvíte, jak pomocí Binderu služby Spring Cloud Stream Event Hubs odesílat zprávy a přijímat je ze služby Event Hubs.

Přidejte závislosti

Pokud chcete nainstalovat modul Spring Cloud Azure Event Hubs Stream Binder, přidejte do souboru pom.xml následující závislosti:

• Spring Cloud Azure 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>
  

  Poznámka:

  Pokud používáte Spring Boot 4.0.x, nastavte spring-cloud-azure-dependencies verzi na 7.4.0hodnotu .

  Pokud používáte Spring Boot 3.5.x, nastavte spring-cloud-azure-dependencies verzi na 6.5.0hodnotu .

  Pokud používáte Spring Boot 3.1.x-3.5.x, nezapomeňte nastavit verzi na spring-cloud-azure-dependencies.

  Pokud používáte Spring Boot 2.x, nezapomeňte nastavit spring-cloud-azure-dependencies verzi na 4.20.0.

  Tato faktura materiálu (BOM) by měla být nakonfigurována v <dependencyManagement> části vašeho pom.xml souboru. Tím zajistíte, že všechny závislosti Spring Cloud Azure budou používat stejnou verzi.

  Další informace o verzi použité pro tento kusovník najdete v tématu Kterou verzi Spring Cloud Azure bych měl použít.

• Artefakt Spring Cloud Azure Event Hubs Stream Binderu:

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

Kódování aplikace pro odesílání a příjem zpráv

Pokud chcete odesílat zprávy do služby Event Hubs a přijímat je, nakonfigurujte aplikaci pomocí následujícího postupu:

1. Ke konfiguraci namespace Event Hubs a Storage Blob použijte následující nastavení vlastností:

   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. Vytvořte příjemce zprávy.

   Pokud chcete aplikaci použít jako zachytávač událostí, nakonfigurujte vstupní propojovač a postupujte podle těchto kroků:

   • Deklarujte Consumer bean, který definuje logiku zpracování zpráv. Například následující Consumer bean má název consume:

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

   • Přidejte následující konfiguraci, abyste zadali Event Hub název pro spotřebu. Nezapomeňte zástupný symbol nahradit skutečnou <your-event-hub-name> hodnotou.

     # 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. Vytvořte odesílatele zprávy.

   Pokud chcete aplikaci použít jako zdroj událostí, nakonfigurujte výstupní pořadač provedením následujících úloh:

   • Definujte Supplier bean, který určuje, odkud ve vaší aplikaci pocházejí zprávy, jak je znázorněno v následujícím příkladu:

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

   • Přidejte následující konfiguraci a zadejte Event Hub název pro odeslání. Nezapomeňte zástupný symbol nahradit skutečnou <your-event-hub-name> hodnotou.

     # "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. Spusťte aplikaci. Zobrazí se protokoly podobné následujícímu příkladu:

   Sending a message.
   New message received: Hello world.
   

Použití Spring Kafka s Azure Event Hubs

Služba Event Hubs poskytuje koncový bod Kafka, který můžou používat vaše stávající aplikace založené na platformě Kafka. Tento přístup nabízí alternativu ke spuštění vlastního clusteru Kafka. Event Hubs funguje s mnoha vašimi existujícími aplikacemi Kafka. Další informace najdete v tématu Event Hubs pro Apache Kafka.

V této příručce se dozvíte, jak používat Azure Event Hubs a Spring Kafka k odesílání a přijímání zpráv ze služby Event Hubs.

Přidejte závislosti

Pokud chcete nainstalovat startovací balíčky Spring Cloud Azure a moduly Spring Kafka, přidejte do souboru pom.xml následující závislosti:

• Spring Cloud Azure 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>
  

  Poznámka:

  Pokud používáte Spring Boot 4.0.x, nastavte spring-cloud-azure-dependencies verzi na 7.4.0hodnotu .

  Pokud používáte Spring Boot 3.5.x, nastavte spring-cloud-azure-dependencies verzi na 6.5.0hodnotu .

  Pokud používáte Spring Boot 3.1.x-3.5.x, nezapomeňte nastavit verzi na spring-cloud-azure-dependencies.

  Pokud používáte Spring Boot 2.x, nezapomeňte nastavit spring-cloud-azure-dependencies verzi na 4.20.0.

  Tato faktura materiálu (BOM) by měla být nakonfigurována v <dependencyManagement> části vašeho pom.xml souboru. Tím zajistíte, že všechny závislosti Spring Cloud Azure budou používat stejnou verzi.

  Další informace o verzi použité pro tento kusovník najdete v tématu Kterou verzi Spring Cloud Azure bych měl použít.

• Úvodní Azure Spring Cloudu a artefakt Spring Kafka:

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

Kódování aplikace pro odesílání a příjem zpráv

Pokud chcete odesílat zprávy do služby Event Hubs a přijímat je, nakonfigurujte aplikaci pomocí následujícího postupu:

1. Ke konfiguraci oboru názvů Event Hubs použijte následující nastavení vlastností:

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

2. Slouží KafkaTemplate k odesílání zpráv a @KafkaListener přijímání zpráv, jak je znázorněno v následujícím příkladu. Nezapomeňte zástupný symbol nahradit skutečnou <your-event-hub-name> hodnotou.

   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. Spusťte aplikaci. Zobrazí se protokoly podobné následujícímu příkladu:

   Message was sent successfully.
   New message received: Hello world
   

Použití Spring Cloud Stream Kafka Binderu s Azure Event Hubs

Spring Cloud Stream je architektura, která vývojářům aplikací umožňuje psát mikroslužby řízené zprávami. Mostem mezi systémem zasílání zpráv a Spring Cloud Stream je vazbová abstrakce. Bindery existují pro několik systémů zasílání zpráv, ale jeden z nejčastěji používaných pořadačů je pro Apache Kafka.

V této příručce se dozvíte, jak používat Azure Event Hubs a Spring Cloud Stream Kafka Binder k odesílání zpráv do a přijímání zpráv ze služby Event Hubs.

Přidejte závislosti

Chcete-li nainstalovat Spring Cloud Azure starter a Spring Cloud Stream Kafka binder moduly, přidejte do souboru pom.xml následující závislosti:

• Spring Cloud Azure 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>
  

  Poznámka:

  Pokud používáte Spring Boot 4.0.x, nastavte spring-cloud-azure-dependencies verzi na 7.4.0hodnotu .

  Pokud používáte Spring Boot 3.5.x, nastavte spring-cloud-azure-dependencies verzi na 6.5.0hodnotu .

  Pokud používáte Spring Boot 3.1.x-3.5.x, nezapomeňte nastavit verzi na spring-cloud-azure-dependencies.

  Pokud používáte Spring Boot 2.x, nezapomeňte nastavit spring-cloud-azure-dependencies verzi na 4.20.0.

  Tato faktura materiálu (BOM) by měla být nakonfigurována v <dependencyManagement> části vašeho pom.xml souboru. Tím zajistíte, že všechny závislosti Spring Cloud Azure budou používat stejnou verzi.

  Další informace o verzi použité pro tento kusovník najdete v tématu Kterou verzi Spring Cloud Azure bych měl použít.

• Úvodní artefakt Azure Spring Cloudu:

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

Kódování aplikace pro odesílání a příjem zpráv

Pokud chcete odesílat zprávy do služby Event Hubs a přijímat je, nakonfigurujte aplikaci pomocí následujícího postupu:

1. Ke konfiguraci zprostředkovatele Kafka použijte následující nastavení vlastnosti:

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

2. Vytvořte příjemce zprávy.

   Pokud chcete aplikaci použít jako zachytávač událostí, nakonfigurujte vstupní propojovač a postupujte podle těchto kroků:

   • Deklarujte Consumer bean, který definuje logiku zpracování zpráv. Například následující Consumer bean má název consume:

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

   • Přidejte následující konfiguraci, abyste zadali Event Hub název pro spotřebu. Nezapomeňte zástupný symbol nahradit skutečnou <your-event-hub-name> hodnotou.

     # 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. Vytvořte odesílatele zprávy.

   Pokud chcete aplikaci použít jako zdroj událostí, nakonfigurujte výstupní pořadač provedením následujících úloh:

   • Definujte Supplier bean, který určuje, odkud ve vaší aplikaci pocházejí zprávy, jak je znázorněno v následujícím příkladu:

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

   • Přidejte následující konfiguraci a zadejte Event Hub název pro odeslání. Nezapomeňte zástupný symbol nahradit skutečnou <your-event-hub-name> hodnotou.

     # "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. Spusťte aplikaci. Zobrazí se protokoly podobné následujícímu příkladu:

   Sending a message.
   New message received: Hello world.
   

Nasazení do Azure Spring Apps

Teď, když máte aplikaci Spring Boot spuštěnou místně, je čas ji přesunout do produkčního prostředí. Azure Spring Apps usnadňuje nasazování aplikací Spring Boot do Azure bez jakýchkoli změn kódu. Služba spravuje infrastrukturu aplikací Spring, aby se vývojáři mohli soustředit na svůj kód. Azure Spring Apps poskytuje správu životního cyklu pomocí komplexního monitorování a diagnostiky, správy konfigurace, zjišťování služeb, integrace CI/CD, modrých zelených nasazení a dalších. Pokud chcete aplikaci nasadit do Azure Spring Apps, viz Nasazení první aplikace na Azure Spring Apps.

Další kroky

Azure pro vývojáře SpringSpring Cloud Azure Event Hubs Samples

Viz také

Další informace o dalších úvodních aplikacích Spring Boot, které jsou k dispozici pro Microsoft Azure, najdete v tématu Co je Spring Cloud Azure?