Udostępnij za pośrednictwem

Usługa Spring Cloud Stream z usługą Azure Event Hubs

  • Artykuł

W tym samouczku pokazano, jak wysyłać i odbierać komunikaty przy użyciu usług Azure Event Hubs i Spring Cloud Stream Binder Eventhubs w aplikacji Spring Boot.

Wymagania wstępne

Uwaga

Aby przyznać swojemu kontu dostęp do zasobów, w usłudze Azure Event Hubs przypisz Azure Event Hubs Data Receiver rolę i Azure Event Hubs Data Sender do aktualnie używanego konta Microsoft Entra. Następnie na koncie usługi Azure Storage przypisz Storage Blob Data Contributor rolę do aktualnie używanego konta Microsoft Entra. Aby uzyskać więcej informacji na temat udzielania ról dostępu, zobacz Przypisywanie ról platformy Azure przy użyciu witryny Azure Portal i Autoryzowanie dostępu do zasobów usługi Event Hubs przy użyciu identyfikatora Entra firmy Microsoft.

Ważne

Do wykonania kroków opisanych w tym samouczku jest wymagana wersja 2.5 lub nowsza.

Wysyłanie i odbieranie komunikatów z usługi Azure Event Hubs

Za pomocą konta usługi Azure Storage i centrum zdarzeń platformy Azure można wysyłać i odbierać komunikaty przy użyciu usługi Event Hubs powiązania usługi Azure Stream Spring Cloud.

Aby zainstalować moduł spring Cloud Azure Stream Binder Event Hubs, dodaj następujące zależności do pliku pom.xml :

  • Projekt Spring Cloud Azure Bill of Materials (BOM):

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

    Uwaga

    Jeśli używasz środowiska Spring Boot 2.x, pamiętaj, aby ustawić spring-cloud-azure-dependencies wersję na 4.19.0. Ten rachunek materiału (BOM) należy skonfigurować w <dependencyManagement> sekcji pliku pom.xml . Gwarantuje to, że wszystkie zależności platformy Azure platformy Spring Cloud korzystają z tej samej wersji. Aby uzyskać więcej informacji na temat wersji używanej dla tego modelu BOM, zobacz Która wersja platformy Spring Cloud platformy Azure powinna być używana.

  • Artefakt usługi Event Hubs powiązania usługi Azure Stream Spring Cloud:

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

Kodowanie aplikacji

Wykonaj poniższe kroki, aby skonfigurować aplikację do tworzenia i używania komunikatów przy użyciu usługi Azure Event Hubs.

  1. Skonfiguruj poświadczenia centrum zdarzeń, dodając następujące właściwości do pliku application.properties .

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

    W poniższej tabeli opisano pola w konfiguracji:

    Pole opis
    spring.cloud.azure.eventhubs.namespace Określ przestrzeń nazw uzyskaną w centrum zdarzeń w witrynie Azure Portal.
    spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name Określ konto magazynu utworzone w tym samouczku.
    spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name Określ kontener konta magazynu.
    spring.cloud.stream.bindings.consume-in-0.destination Określ centrum zdarzeń używane w tym samouczku.
    spring.cloud.stream.bindings.consume-in-0.group Określ grupy odbiorców w wystąpieniu usługi Event Hubs.
    spring.cloud.stream.bindings.supply-out-0.destination Określ to samo centrum zdarzeń, które zostało użyte w tym samouczku.
    spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode Podaj wartość MANUAL.
    spring.cloud.function.definition Określ, która fasola funkcjonalna ma być powiązana z zewnętrznymi miejscami docelowymi udostępnianymi przez powiązania.
    spring.cloud.stream.poller.initial-delay Określ początkowe opóźnienie dla wyzwalaczy okresowych. Wartość domyślna to 0.
    spring.cloud.stream.poller.fixed-delay Określ stałe opóźnienie domyślnegollera w milisekundach. Wartość domyślna to 1000 L.

  2. Edytuj plik klasy uruchamiania, aby wyświetlić następującą zawartość.

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

}

    Napiwek

    W tym samouczku nie ma żadnych operacji uwierzytelniania w konfiguracjach ani kodzie. Jednak nawiązywanie połączenia z usługami platformy Azure wymaga uwierzytelniania. Aby ukończyć uwierzytelnianie, musisz użyć usługi Azure Identity. Platforma Spring Cloud na platformie Azure używa DefaultAzureCredentialbiblioteki tożsamości platformy Azure, która ułatwia uzyskiwanie poświadczeń bez żadnych zmian w kodzie.

    DefaultAzureCredential obsługuje wiele metod uwierzytelniania i określa, która metoda ma być używana w czasie wykonywania. Takie podejście umożliwia aplikacji używanie różnych metod uwierzytelniania w różnych środowiskach (takich jak środowiska lokalne i produkcyjne) bez implementowania kodu specyficznego dla środowiska. Aby uzyskać więcej informacji, zobacz DefaultAzureCredential.

    Aby ukończyć uwierzytelnianie w lokalnych środowiskach deweloperskich, możesz użyć interfejsu wiersza polecenia platformy Azure, programu Visual Studio Code, programu PowerShell lub innych metod. Aby uzyskać więcej informacji, zobacz Uwierzytelnianie platformy Azure w środowiskach deweloperskich Java. Aby ukończyć uwierzytelnianie w środowiskach hostingu platformy Azure, zalecamy użycie tożsamości zarządzanej przypisanej przez użytkownika. Aby uzyskać więcej informacji, zobacz Co to są tożsamości zarządzane dla zasobów platformy Azure?

  3. Uruchom aplikację. Komunikaty podobne do tych będą publikowane w dzienniku aplikacji, jak pokazano w poniższych przykładowych danych wyjściowych:

    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

Wdrażanie w usłudze Azure Spring Apps

Teraz, gdy aplikacja Spring Boot działa lokalnie, nadszedł czas, aby przenieść ją do środowiska produkcyjnego. Usługa Azure Spring Apps ułatwia wdrażanie aplikacji Spring Boot na platformie Azure bez żadnych zmian w kodzie. Usługa zarządza infrastrukturą aplikacji Spring, aby deweloperzy mogli skupić się na swoim kodzie. Usługa Azure Spring Apps zapewnia zarządzanie cyklem życia przy użyciu kompleksowego monitorowania i diagnostyki, zarządzania konfiguracją, odnajdywania usług, integracji/ciągłego wdrażania, wdrożeń niebieskich zielonych i nie tylko. Aby wdrożyć aplikację w usłudze Azure Spring Apps, zobacz Wdrażanie pierwszej aplikacji w usłudze Azure Spring Apps.

Następne kroki

Platforma Azure dla deweloperów platformySpring Cloud — przykłady usługi Event Hubs binder usługi Azure Stream