Az Event Hubs használata Spring-alkalmazásokban

Ez a cikk bemutatja, hogyan használhatja a Azure Event Hubs Java Spring-keretrendszerrel készült alkalmazásokban.

Azure Event Hubs egy big data streamplatform és eseménybetöltési szolgáltatás. Másodpercenként több millió eseményt képes fogadni és feldolgozni. Az eseményközpontokba elküldött adatok bármilyen valós idejű elemzési szolgáltató vagy kötegelési/tárolóadapter segítségével átalakíthatók és tárolhatók.

A Spring Cloud Azure különböző modulokat biztosít az Event Hubsba érkező és fogadott üzenetekhez a Spring-keretrendszerek használatával.

A következő modulokat egymástól függetlenül használhatja, vagy kombinálhatja őket különböző használati esetekhez:

Előfeltételek

Feljegyzés

Ha hozzáférést szeretne adni a fiókjának az erőforrásokhoz, az Azure Event Hubsban rendelje az Azure Event Hubs Data Receiver és Azure Event Hubs Data Sender szerepköröket az éppen használt Microsoft Entra fiókhoz. Ezután a Azure Storage fiókban rendelje hozzá a Storage Blob Data Contributor szerepkört az éppen használt Microsoft Entra fiókhoz. A hozzáférési szerepkörök engedélyezésével kapcsolatos további információkért lásd: Az Azure szerepkörök hozzárendelése a Azure portálon és Az Event Hubs-erőforrásokhoz való hozzáférés engedélyezése Microsoft Entra ID használatával.

Fontos

Az oktatóanyag lépéseinek elvégzéséhez a Spring Boot 2.5-ös vagy újabb verziójára van szükség.

A helyi környezet előkészítése

Ebben az oktatóanyagban a konfigurációk és a kód nem rendelkeznek hitelesítési műveletekkel. Az Azure szolgáltatáshoz való csatlakozáshoz azonban hitelesítés szükséges. A hitelesítés befejezéséhez a Azure Identity ügyfélkódtárat kell használnia. A Spring Cloud Azure DefaultAzureCredential használ, amelyet az Azure Identitástár biztosít a hitelesítő adatok kódmódosítások nélküli beszerzéséhez.

DefaultAzureCredential több hitelesítési módszert támogat, és meghatározza, hogy melyik metódust használja futásidőben. Ez a megközelítés lehetővé teszi, hogy az alkalmazás különböző hitelesítési módszereket használjon különböző környezetekben – például helyi vagy éles környezetekben – környezetspecifikus kód implementálása nélkül. További információért lásd a DefaultAzureCredential szakaszt az Authenticate Azure által üzemeltetett Java alkalmazások című részben.

Ha Azure CLI, IntelliJ vagy más módszereket szeretne használni a helyi fejlesztési környezetekben történő hitelesítés befejezéséhez, tekintse meg Azure hitelesítést Java fejlesztési környezetekben. A hitelesítés Azure üzemeltetési környezetekben való befejezéséhez ajánlott felügyelt identitást használni. További információ: Mi az Azure-erőforrások felügyelt identitásai?

Spring Cloud Azure Event Hubs Starter használata

A Spring Cloud Azure Event Hubs Starter modul importálja a Event Hubs Java ügyfélkódtárat a Spring Boot keretrendszerrel. A Spring Cloud Azure-t és az Azure SDK-t együtt használhatja úgy, hogy az nem kölcsönösen kizárólagos. Így továbbra is használhatja az Event Hubs Java ügyfél API-t a Spring-alkalmazásban.

Függőségek hozzáadása

A Spring Cloud Azure Event Hubs Starter modul telepítéséhez adja hozzá a következő függőségeket a pom.xml fájlhoz:

  • A Spring Cloud Azure Anyagjegyzék (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>

    Feljegyzés

    Ha Spring Boot 4.0.x-et használ, állítsa a verziót a spring-cloud-azure-dependencies következőre 7.4.0: .

    Ha a Spring Boot 3.5.x-et használja, állítsa a verziót a spring-cloud-azure-dependencies következőre 6.5.0: .

    Ha a Spring Boot 3.1.x-3.5.x verziót használja, mindenképpen állítsa be a verziót spring-cloud-azure-dependencies-ra/c1.

    Ha a Spring Boot 2.x-et használja, mindenképpen állítsa be a spring-cloud-azure-dependencies verziót 4.20.0-re.

    Ezt az anyagjegyzéket (BOM) a <dependencyManagement> fájl szakaszában kell konfigurálni. Ez biztosítja, hogy minden Spring Cloud-Azure függőség ugyanazt a verziót használja.

    Az anyagjegyzékhez használt verzióval kapcsolatos további információkért lásd: Melyik verzióját a Spring Cloud Azure-nak érdemes használnom?.

  • A Spring Cloud Azure Event Hubs összetevő:

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

Az alkalmazás kódolása üzenetek küldéséhez és fogadásához

Ez az útmutató bemutatja, hogyan használhatja az Event Hubs Java-ügyfeleket egy Spring-alkalmazás kontextusában. Itt a következő két lehetőséget mutatjuk be:

  • Használja a Spring Boot automatikus konfigurálását, és használja a Spring-környezetből származó beépített ügyfeleket (ajánlott).
  • Hozza létre a klienst programozás útján.

A Spring IoC konténerből származó kliens bean-ek automatikus kábelezésének módja a következő előnyökkel jár, amelyek rugalmasabb és hatékonyabb élményt nyújtanak a fejlesztési folyamat során az Event Hubs ügyfelekkel.

  • A külső konfigurációt alkalmazza, hogy ugyanazt az alkalmazáskódot használhassa különböző környezetekben.
  • A Spring Boot-keretrendszerbe delegálhatja a szerkesztői minta megismerésének és az ügyfél alkalmazáskörnyezetbe való regisztrálásának folyamatát. Ezzel a delegálással arra összpontosíthat, hogyan használhatja az ügyfeleket saját üzleti igényei szerint.
  • Az állapotjelző segítségével egyszerűen megvizsgálhatja az alkalmazás és a belső összetevők állapotát és állapotát.

Az alábbi szakaszok olyan kódpéldákat mutatnak be, amelyek megmutatják, hogyan használhatja a EventProcessorClient és EventHubProducerClient a két alternatíva esetében.

Feljegyzés

Azure Java Event Hubshoz készült SDK több ügyfelet is biztosít az Event Hubs használatához. A starter automatikus konfigurálást is biztosít az összes Event Hubs-ügyfélhez, valamint az ügyfélkészítőkhöz. Ez a cikk csak EventProcessorClientEventHubProducerClient példákat használ.

A Spring Boot automatikus konfigurációjának használata

Ha üzeneteket szeretne küldeni és fogadni az Event Hubstól, konfigurálja az alkalmazást az alábbi lépések végrehajtásával:

  1. Az Event Hubs-névtér és az eseményközpont nevének konfigurálásához használja a következő tulajdonságbeállításokat:

    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. Hozzon létre egy új EventHubProcessorClientConfiguration Java osztályt az alábbi példában látható módon. Ez az osztály az üzenet és a hibakezelő regisztrálásához EventProcessorClienthasználható.

    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. Injektálja a EventProcessorClient és EventHubProducerClient elemeket a Spring-alkalmazásába, majd hívja meg a kapcsolódó API-kat üzenetek küldéséhez és fogadásához, ahogyan az a következő példában látható.

    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. Indítsa el az alkalmazást. A következő példához hasonló naplók jelennek meg:

    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.

Az ügyfél programozott módon történő létrehozása

Az ügyfélkomponenseket önállóan is létrehozhatja, de a folyamat bonyolult. A Spring Boot-alkalmazásokban kezelnie kell a tulajdonságokat, meg kell tanulnia a szerkesztői mintát, és regisztrálnia kell az ügyfelet a Spring-alkalmazás környezetében. A következő lépések bemutatják, hogyan teheti ezt meg:

  1. Hozzon létre egy új EventHubClientConfiguration Java osztályt az alábbi példában látható módon. Ez az osztály a EventProcessorClient és EventHubProducerClient bean-ek deklarálásához használatos. Mindenképpen cserélje le a <your event-hubs-namespace>, <your-event-hub-name>, <your-storage-account-name>és <your-storage-account-container-name> helyőrzőket a tényleges értékekre.

    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. Injektálja a EventProcessorClient és EventHubProducerClient elemeket a Spring alkalmazásába, ahogy az az alábbi példában látható:

    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. Indítsa el az alkalmazást. A következő példához hasonló naplók jelennek meg:

    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.

Az alábbi listában látható néhány ok, amiért ez a kód nem rugalmas vagy kecses:

  • Az Event Hubs-névtér és az eseményközpont neve kódolt.
  • Ha a Spring-környezetből szeretne @Value konfigurációkat lekérni, az application.properties fájlban nem szerepelhetnek IDE-tippek.
  • Ha mikroszolgáltatási forgatókönyve van, minden projektben meg kell duplikálnia a kódot, és könnyen hibázhat, és nehéz konzisztensnek lennie.

Szerencsére a Spring Cloud Azure használatával nem szükséges önállóan felépíteni az ügyfélkomponenseket. Ehelyett közvetlenül injektálhatja őket, és használhatja azokat a konfigurációs tulajdonságokat, amelyeket már ismer a Storage-üzenetsor konfigurálásához. További információ: Spring Cloud Azure konfiguráció.

A Spring Cloud Azure a következő globális konfigurációkat is biztosítja a különböző forgatókönyvekhez. További információért lásd a Spring Cloud Azure konfiguráció című rész Globális konfiguráció Azure Service SDK-khoz szakaszát.

  • Proxybeállítások.
  • Újrapróbálkozás beállításai.
  • AMQP átviteli ügyfél beállításai.

Különböző Azure felhőkhöz is csatlakozhat. További információ: Kapcsolódás különböző Azure felhőkhöz.

Spring Messaging Azure Event Hubs használata

A Spring Messaging Azure Event Hubs modul támogatja Spring Messaging keretrendszert az Event Hubs használatával.

Ha Spring Messaging Azure Event Hubs használ, az alábbi funkciókat használhatja:

  • EventHubsTemplate: Üzenetek küldése Eseményközpontokba aszinkron és szinkron módon.
  • @EventHubsListener: Jelöljön meg egy metódust, amely egy Event Hubs-üzenetfigyelő célpontja a célhelyen.

Ez az útmutató bemutatja, hogyan használhatja a Spring Messaging Azure Event Hubs az Event Hubs-üzenetek küldésére és fogadására.

Függőségek hozzáadása

A Spring Messaging Azure Event Hubs modul telepítéséhez adja hozzá a következő függőségeket a pom.xml fájlhoz:

  • A Spring Cloud Azure Anyagjegyzék (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>

    Feljegyzés

    Ha Spring Boot 4.0.x-et használ, állítsa a verziót a spring-cloud-azure-dependencies következőre 7.4.0: .

    Ha a Spring Boot 3.5.x-et használja, állítsa a verziót a spring-cloud-azure-dependencies következőre 6.5.0: .

    Ha a Spring Boot 3.1.x-3.5.x verziót használja, mindenképpen állítsa be a verziót spring-cloud-azure-dependencies-ra/c1.

    Ha a Spring Boot 2.x-et használja, mindenképpen állítsa be a spring-cloud-azure-dependencies verziót 4.20.0-re.

    Ezt az anyagjegyzéket (BOM) a <dependencyManagement> fájl szakaszában kell konfigurálni. Ez biztosítja, hogy minden Spring Cloud-Azure függőség ugyanazt a verziót használja.

    Az anyagjegyzékhez használt verzióval kapcsolatos további információkért lásd: Melyik verzióját a Spring Cloud Azure-nak érdemes használnom?.

  • A Spring Cloud Azure starter, Spring Messaging Event Hubs és Azure Event Hubs Checkpoint Store-összetevők:

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

Az alkalmazás kódolása üzenetek küldéséhez és fogadásához

Ha üzeneteket szeretne küldeni és fogadni az Event Hubstól, konfigurálja az alkalmazást az alábbi lépések végrehajtásával:

  1. Az Event Hubs-névtér és a Storage-blob konfigurálásához használja a következő tulajdonságbeállításokat:

    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. Hozzon létre egy új ConsumerService Java osztályt az alábbi példában látható módon. Ez az osztály egy üzenet fogadó definiálására szolgál. Ügyeljen arra, hogy a <your-event-hub-name> helyőrzőt a tényleges értékére cserélje le.

    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. Küldjön egy feladót és egy fogadót, hogy üzeneteket küldjön és fogadjon a Spring használatával, ahogyan az az alábbi példában látható. Ügyeljen arra, hogy a <your-event-hub-name> helyőrzőt a tényleges értékére cserélje le.

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

}

    Tipp.

    Ne felejtse el hozzáadni a @EnableAzureMessaging annotációt, amely elindítja a @EventHubsListener-el megjelölt metódusok felderítését, és létrehozza az üzenetfigyelő tárolót a háttérben.

  4. Indítsa el az alkalmazást. A következő példához hasonló naplók jelennek meg:

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

A Spring Integration Azure Event Hubs használata

Az Azure Event Hubs Spring integrációs modul támogatást nyújt a Spring Integration keretrendszer számára az Event Hubs használatával.

Ha a Spring-alkalmazás Spring Integration üzenetcsatornákat használ, csatornaadapterekkel irányíthatja az üzeneteket az üzenetcsatornák és az Event Hubs között.

A bejövő csatornaadapter üzeneteket továbbít egy eseményközpontból egy üzenetcsatornába. A kimenő csatornaadapter üzeneteket tesz közzé egy üzenetcsatornából egy eseményközpontba.

Ez az útmutató bemutatja, hogyan küldhet üzeneteket az Event Hubsnak a Spring Integration Azure Event Hubs használatával.

Függőségek hozzáadása

A Spring Cloud Azure Event Hubs Integration Starter modul telepítéséhez adja hozzá a következő függőségeket a pom.xml fájlhoz:

  • A Spring Cloud Azure Anyagjegyzék (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>

    Feljegyzés

    Ha Spring Boot 4.0.x-et használ, állítsa a verziót a spring-cloud-azure-dependencies következőre 7.4.0: .

    Ha a Spring Boot 3.5.x-et használja, állítsa a verziót a spring-cloud-azure-dependencies következőre 6.5.0: .

    Ha a Spring Boot 3.1.x-3.5.x verziót használja, mindenképpen állítsa be a verziót spring-cloud-azure-dependencies-ra/c1.

    Ha a Spring Boot 2.x-et használja, mindenképpen állítsa be a spring-cloud-azure-dependencies verziót 4.20.0-re.

    Ezt az anyagjegyzéket (BOM) a <dependencyManagement> fájl szakaszában kell konfigurálni. Ez biztosítja, hogy minden Spring Cloud-Azure függőség ugyanazt a verziót használja.

    Az anyagjegyzékhez használt verzióval kapcsolatos további információkért lásd: Melyik verzióját a Spring Cloud Azure-nak érdemes használnom?.

  • A Spring Cloud Azure Event Hubs integrációs összetevő:

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

Az alkalmazás kódolása üzenetek küldéséhez és fogadásához

Ha üzeneteket szeretne küldeni és fogadni az Event Hubstól, konfigurálja az alkalmazást az alábbi lépések végrehajtásával:

  1. Az Event Hubs-névtér és a Storage-blob konfigurálásához használja a következő tulajdonságbeállításokat:

    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. Hozzon létre egy új MessageReceiveConfiguration Java osztályt az alábbi példában látható módon. Ez az osztály egy üzenet fogadó definiálására szolgál. Ügyeljen arra, hogy a <your-event-hub-name> helyőrzőt a tényleges értékére cserélje le.

    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. Hozzon létre egy új MessageSendConfiguration Java osztályt az alábbi példában látható módon. Ez az osztály egy üzenetküldő definiálására szolgál. Ügyeljen arra, hogy a <your-event-hub-name> helyőrzőt a tényleges értékére cserélje le.

    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. Küldjön egy feladót és egy fogadót, hogy üzeneteket küldjön és fogadjon a Spring használatával, ahogy az a következő példában látható:

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

    Tipp.

    Ne felejtse el hozzáadni a @EnableIntegration megjegyzést, amely lehetővé teszi a Spring Integration-infrastruktúrát.

  5. Indítsa el az alkalmazást. A következő példához hasonló naplók jelennek meg:

    Message was sent successfully.
New message received: Hello World

A Spring Cloud Azure Stream Event Hubs Binder használata

Az Event Hubs API meghívásához egy Spring Cloud Stream alkalmazásban használja a Spring Cloud Azure Event Hubs Stream Binder modult.

Ez az útmutató bemutatja, hogyan küldhet üzeneteket az Event Hubsnak a Spring Cloud Stream Event Hubs Binder használatával.

Függőségek hozzáadása

A Spring Cloud Azure Event Hubs Stream Binder modul telepítéséhez adja hozzá a következő függőségeket a pom.xml fájlhoz:

  • A Spring Cloud Azure Anyagjegyzék (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>

    Feljegyzés

    Ha Spring Boot 4.0.x-et használ, állítsa a verziót a spring-cloud-azure-dependencies következőre 7.4.0: .

    Ha a Spring Boot 3.5.x-et használja, állítsa a verziót a spring-cloud-azure-dependencies következőre 6.5.0: .

    Ha a Spring Boot 3.1.x-3.5.x verziót használja, mindenképpen állítsa be a verziót spring-cloud-azure-dependencies-ra/c1.

    Ha a Spring Boot 2.x-et használja, mindenképpen állítsa be a spring-cloud-azure-dependencies verziót 4.20.0-re.

    Ezt az anyagjegyzéket (BOM) a <dependencyManagement> fájl szakaszában kell konfigurálni. Ez biztosítja, hogy minden Spring Cloud-Azure függőség ugyanazt a verziót használja.

    Az anyagjegyzékhez használt verzióval kapcsolatos további információkért lásd: Melyik verzióját a Spring Cloud Azure-nak érdemes használnom?.

  • A Spring Cloud Azure Event Hubs Stream Binder kötőelem:

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

Az alkalmazás kódolása üzenetek küldéséhez és fogadásához

Ha üzeneteket szeretne küldeni és fogadni az Event Hubstól, konfigurálja az alkalmazást az alábbi lépések végrehajtásával:

  1. Az Event Hubs-névtér és a Storage-blob konfigurálásához használja a következő tulajdonságbeállításokat:

    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. Hozza létre az üzenet fogadót.

    Az alkalmazás eseménygyűjtőként való használatához konfigurálja a bemeneti iratgyűjtőt az alábbi feladatok elvégzésével:

    • Deklaráljon egy Consumer bean-t, amely meghatározza az üzenetkezelési logikát. A Consumer bab neve például consume:

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

    • Adja hozzá a következő konfigurációt a Event Hub fogyasztási nevének megadásához. Ügyeljen arra, hogy a <your-event-hub-name> helyőrzőt a tényleges értékére cserélje le.

      # 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. Hozza létre az üzenet küldőjét.

    Az alkalmazás eseményforrásként való használatához konfigurálja a kimeneti kötést a következő feladatok végrehajtásával:

    • Határozza meg a Supplier bean-t, amely meghatározza, hogy az üzenetek honnan érkeznek az alkalmazáson belül, ahogyan az alábbi példában látható:

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

    • Adja hozzá a következő konfigurációt a Event Hub küldés nevének megadásához. Ügyeljen arra, hogy a <your-event-hub-name> helyőrzőt a tényleges értékére cserélje le.

      # "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. Indítsa el az alkalmazást. A következő példához hasonló naplók jelennek meg:

    Sending a message.
New message received: Hello world.

Azure Event Hubs használata Spring Kaffkával

Az Event Hubs egy Olyan Kafka-végpontot biztosít, amelyet a meglévő Kafka-alapú alkalmazások használhatnak. Ez a megközelítés alternatívát kínál a saját Kafka-fürt üzemeltetésére. Az Event Hubs számos meglévő Kafka-alkalmazással működik együtt. További információ: Event Hubs for Apache Kafka.

Ez az útmutató bemutatja, hogyan használhatja Azure Event Hubs és Spring Kafka használatával üzeneteket küldeni és fogadni az Event Hubstól.

Függőségek hozzáadása

A Spring Cloud Azure starter- és Spring Kafka-modulok telepítéséhez adja hozzá a következő függőségeket a pom.xml fájlhoz:

  • A Spring Cloud Azure Anyagjegyzék (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>

    Feljegyzés

    Ha Spring Boot 4.0.x-et használ, állítsa a verziót a spring-cloud-azure-dependencies következőre 7.4.0: .

    Ha a Spring Boot 3.5.x-et használja, állítsa a verziót a spring-cloud-azure-dependencies következőre 6.5.0: .

    Ha a Spring Boot 3.1.x-3.5.x verziót használja, mindenképpen állítsa be a verziót spring-cloud-azure-dependencies-ra/c1.

    Ha a Spring Boot 2.x-et használja, mindenképpen állítsa be a spring-cloud-azure-dependencies verziót 4.20.0-re.

    Ezt az anyagjegyzéket (BOM) a <dependencyManagement> fájl szakaszában kell konfigurálni. Ez biztosítja, hogy minden Spring Cloud-Azure függőség ugyanazt a verziót használja.

    Az anyagjegyzékhez használt verzióval kapcsolatos további információkért lásd: Melyik verzióját a Spring Cloud Azure-nak érdemes használnom?.

  • A Spring Cloud Azure starter és a Spring Kafka összetevő:

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

Az alkalmazás kódolása üzenetek küldéséhez és fogadásához

Ha üzeneteket szeretne küldeni és fogadni az Event Hubstól, konfigurálja az alkalmazást az alábbi lépések végrehajtásával:

  1. Az Event Hubs-névtér konfigurálásához használja a következő tulajdonságbeállítást:

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

  2. Az KafkaTemplate alábbi példában látható módon üzeneteket küldhet és @KafkaListener fogadhat. Ügyeljen arra, hogy a <your-event-hub-name> helyőrzőt a tényleges értékére cserélje le.

    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. Indítsa el az alkalmazást. A következő példához hasonló naplók jelennek meg:

    Message was sent successfully.
New message received: Hello world

A Spring Cloud Stream Kafka Binder használata Azure Event Hubs

A Spring Cloud Stream egy keretrendszer, amely lehetővé teszi az alkalmazásfejlesztők számára az üzenetvezérelt mikroszolgáltatások írását. Az üzenetkezelési rendszer és a Spring Cloud Stream közötti híd a binderek absztrakcióján keresztül történik. A binderek számos üzenetkezelési rendszerhez léteznek, de az egyik leggyakrabban használt kötés az Apache Kafka.

Ez az útmutató bemutatja, hogyan használhatja Azure Event Hubs és Spring Cloud Stream Kafka Binder használatával üzeneteket küldhet az Event Hubsnak, és fogadhat üzeneteket.

Függőségek hozzáadása

A Spring Cloud Azure kezdő- és Spring Cloud Stream kötőanyag Kafka-moduljainak telepítéséhez adja hozzá a következő függőségeket a pom.xml fájlhoz:

  • A Spring Cloud Azure Anyagjegyzék (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>

    Feljegyzés

    Ha Spring Boot 4.0.x-et használ, állítsa a verziót a spring-cloud-azure-dependencies következőre 7.4.0: .

    Ha a Spring Boot 3.5.x-et használja, állítsa a verziót a spring-cloud-azure-dependencies következőre 6.5.0: .

    Ha a Spring Boot 3.1.x-3.5.x verziót használja, mindenképpen állítsa be a verziót spring-cloud-azure-dependencies-ra/c1.

    Ha a Spring Boot 2.x-et használja, mindenképpen állítsa be a spring-cloud-azure-dependencies verziót 4.20.0-re.

    Ezt az anyagjegyzéket (BOM) a <dependencyManagement> fájl szakaszában kell konfigurálni. Ez biztosítja, hogy minden Spring Cloud-Azure függőség ugyanazt a verziót használja.

    Az anyagjegyzékhez használt verzióval kapcsolatos további információkért lásd: Melyik verzióját a Spring Cloud Azure-nak érdemes használnom?.

  • A Spring Cloud Azure kezdő összetevő:

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

Az alkalmazás kódolása üzenetek küldéséhez és fogadásához

Ha üzeneteket szeretne küldeni és fogadni az Event Hubstól, konfigurálja az alkalmazást az alábbi lépések végrehajtásával:

  1. A Kafka-közvetítő konfigurálásához használja a következő tulajdonságbeállítást:

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

  2. Hozza létre az üzenet fogadót.

    Az alkalmazás eseménygyűjtőként való használatához konfigurálja a bemeneti iratgyűjtőt az alábbi feladatok elvégzésével:

    • Deklaráljon egy Consumer bean-t, amely meghatározza az üzenetkezelési logikát. A Consumer bab neve például consume:

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

    • Adja hozzá a következő konfigurációt a Event Hub fogyasztási nevének megadásához. Ügyeljen arra, hogy a <your-event-hub-name> helyőrzőt a tényleges értékére cserélje le.

      # 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. Hozza létre az üzenet küldőjét.

    Az alkalmazás eseményforrásként való használatához konfigurálja a kimeneti kötést a következő feladatok végrehajtásával:

    • Határozza meg a Supplier bean-t, amely meghatározza, hogy az üzenetek honnan érkeznek az alkalmazáson belül, ahogyan az alábbi példában látható:

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

    • Adja hozzá a következő konfigurációt a Event Hub küldés nevének megadásához. Ügyeljen arra, hogy a <your-event-hub-name> helyőrzőt a tényleges értékére cserélje le.

      # "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. Indítsa el az alkalmazást. A következő példához hasonló naplók jelennek meg:

    Sending a message.
New message received: Hello world.

Üzembe helyezés a Azure Spring Apps

Most, hogy a Spring Boot-alkalmazás helyileg fut, ideje éles környezetbe átvinni. Azure Spring Apps megkönnyíti a Spring Boot-alkalmazások Azure való üzembe helyezését kódmódosítások nélkül. A szolgáltatás kezeli a Spring-alkalmazások infrastruktúráját, hogy a fejlesztők a kódjukra összpontosíthassanak. Azure Spring Apps teljes körű monitorozás és diagnosztika, konfigurációkezelés, szolgáltatásfelderítés, CI/CD-integráció, kék-zöld környezetek és egyebek használatával biztosítja az életciklus-felügyeletet. Az alkalmazás Azure Spring Apps való üzembe helyezéséről a A Azure Spring Apps első alkalmazás üzembe helyezéséről szóló cikkben olvashat.

Következő lépések

Azure Spring-fejlesztőknekSpring Cloud Azure Event Hubs minták

Lásd még

További információ a Microsoft Azure elérhető többi Spring Boot Starterről: Mi a Spring Cloud Azure?

  • Utolsó frissítés időpontja: