Spring Cloud Stream az Azure Event Hubsszal

Ez az oktatóanyag bemutatja, hogyan küldhet és fogadhat üzeneteket Azure Event Hubs és Spring Cloud Stream Binder Eventhubs használatával egy Spring Boot-alkalmazásban.

Előfeltételek

Egy Azure előfizetés – ingyenes előfizetés.
Java Fejlesztői készlet (JDK) 8-os vagy újabb verzió.
Apache Maven, 3.2-es vagy újabb verzió.
cURL vagy hasonló HTTP-segédprogram a funkciók teszteléséhez.
Egy Azure Eseményközpont. Ha nincs ilyenje, Azure portál használatával hozhat létre eseményközpontot.
Egy Azure Storage-fiók az Event Hub ellenőrzőpontjaihoz. Ha nem rendelkezik ilyen fiókkal, hozzon létre egy tárfiókot.
Egy Spring Boot-alkalmazás. Ha még nem rendelkezik ilyennel, hozzon létre egy Maven-projektet a Spring Initializrsegítségével. Mindenképpen válassza a Maven Project és a Dependencies területen adja hozzá a Spring web és Azure támogatási függőségeket, majd válassza Java 8-es vagy újabb verzióját.

Feljegyzés

Ha hozzáférést szeretne adni a fiókjának a Azure Event Hubs erőforrásaihoz, rendelje hozzá a Azure Event Hubs Data Receiver szerepköröket az Azure Event Hubs Data Sender é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.

Üzenetek küldése és fogadása az Azure Event Hubs-ból

Egy Azure Storage-fiók és egy Azure Event Hubs használatával üzeneteket küldhet és fogadhat a Spring Cloud Azure Stream Binder Event Hubs használatával.

A Spring Cloud Azure Stream Binder 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 Stream Binder Event Hubs artefaktum:

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

Az alkalmazás kódolása

Az alábbi lépésekkel konfigurálhatja az alkalmazást üzenetek előállítására és felhasználására a Azure Event Hubs használatával.

Konfigurálja az Event Hubs hitelesítő adatait úgy, hogy hozzáadja az alábbi tulajdonságokat az application.properties fájlhoz .

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

Az alábbi táblázat a konfiguráció mezőit ismerteti:

Mező	Leírás
`spring.cloud.azure.eventhubs.namespace`	Adja meg az eseményközponthoz kapott névteret a Azure portálon.
`spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name`	Adja meg a tárfiókot, amelyet ebben az oktatóanyagban létrehozott.
`spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name`	Adja meg a tárfiók tárolóját.
`spring.cloud.stream.bindings.consume-in-0.destination`	Adja meg az oktatóanyagban használt eseményközpontot.
`spring.cloud.stream.bindings.consume-in-0.group`	Adja meg a fogyasztói csoportokat az Event Hubs-példányban.
`spring.cloud.stream.bindings.supply-out-0.destination`	Adja meg ugyanazt az eseményközpontot, amelyet ebben az oktatóanyagban használt.
`spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode`	Adja meg a következőt: `MANUAL`.
`spring.cloud.function.definition`	Adja meg, hogy melyik funkcionális babot szeretné a kötések által közzétett külső célhelyhez kötni.
`spring.cloud.stream.poller.initial-delay`	Adja meg a rendszeres eseményindítók kezdeti késleltetését. Az alapértelmezett érték a `0`.
`spring.cloud.stream.poller.fixed-delay`	Adja meg az alapértelmezett poller rögzített késleltetését ezredmásodpercben. Az alapértelmezett érték a `1000 L`.

Szerkessze az indítási osztályfájlt a következő tartalom megjelenítéséhez.

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

@SpringBootApplication
public class EventHubBinderApplication implements CommandLineRunner {

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

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

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

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

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

}

Tipp.

Ebben az oktatóanyagban nincsenek hitelesítési műveletek a konfigurációkban vagy a kódban. Az Azure szolgáltatásokhoz való csatlakozáshoz azonban hitelesítés szükséges. A hitelesítés befejezéséhez Azure Identitást 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 és éles környezetekben) környezetspecifikus kód implementálása nélkül. További információ: DefaultAzureCredential.

A helyi fejlesztési környezetekben a hitelesítés befejezéséhez használhat Azure CLI, Visual Studio Code, PowerShellt vagy más módszereket. További információ: Azure hitelesítés Java fejlesztési környezetekben. Ha Azure üzemeltetési környezetekben szeretné elvégezni a hitelesítést, javasoljuk, hogy használja a felhasználó által hozzárendelt felügyelt identitást. További információ: Mi az Azure-erőforrások felügyelt identitásai?

Indítsa el az alkalmazást. Az ehhez hasonló üzenetek az alkalmazásnaplóban, az alábbi példakimenetben látható módon:

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

Üzembe helyezés a Azure Spring Apps

Most, hogy a Spring Boot alkalmazás helyileg fut, itt az ideje éles környezetbe helyezni. 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őknek Spring Cloud Azure Stream Binder Event Hubs minták

Visszajelzés

Hasznosnak találta ezt az oldalt?

Last updated on 2026-08-13