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 rolę Azure Event Hubs Data Receiver i Azure Event Hubs Data Sender dla 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 Spring Boot 2.5 lub nowsza.

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

Za pomocą konta usługi Azure Storage i Azure Event Hubs można wysyłać i odbierać komunikaty przy użyciu Spring Cloud Azure Stream Binder Event Hubs.

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.21.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. To zestawienie materiałowe (BOM) należy skonfigurować w <dependencyManagement> sekcji pliku pom.xml. Gwarantuje to, że wszystkie zależności Spring Cloud Azure korzystają z tej samej wersji. Aby uzyskać więcej informacji na temat wersji używanej dla tego BOM, zobacz Która wersja Spring Cloud Azure powinna być używana.

  • Artefakt Spring Cloud Azure Stream Binder dla Event Hubs:

    <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 Event Hub, 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 wydarzeń w portalu Azure.
    spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name Określ konto przechowywania stworzono w tym samouczku.
    spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name Wskaż kontener swojego konta magazynowego.
    spring.cloud.stream.bindings.consume-in-0.destination Określ centrum zdarzeń, którego użyłeś 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órego użyto w tym samouczku.
    spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode Podaj wartość MANUAL.
    spring.cloud.function.definition Określ, który komponent funkcjonalny ma być powiązany z zewnętrznymi celami udostępnianymi przez wią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ślnego pollera 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 konfiguracje i kod nie zawierają żadnych operacji uwierzytelniania. 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 do 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/CD, wdrożeń blue-green 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

Azure dla deweloperów SpringSpring Cloud Azure Stream Binder — przykłady Event Hubs