Spring Cloud Stream con Azure Event Hubs

Questa esercitazione illustra come inviare e ricevere messaggi usando Azure Event Hubs e Spring Cloud Stream Binder Eventhubs in un'applicazione Spring Boot.

Prerequisiti

Una sottoscrizione Azure - crearne una gratuitamente.
Java Development Kit (JDK) versione 8 o successiva.
Apache Maven versione 3.2 o successiva.
cURL o un'utilità HTTP simile per testare la funzionalità.
Un hub eventi Azure. Se non ne hai uno, è necessario creare un hub eventi usando il portale di Azure.
Un account Azure Storage per i checkpoint dell'hub eventi. Se non ne hai uno, crea un account di archiviazione.
Applicazione Spring Boot. Se non ne hai uno, crea un progetto Maven con Spring Initializr. Assicurarsi di selezionare Maven Project e in Dependencies aggiungere le dipendenze Spring Web e Azure Supporto e quindi selezionare Java versione 8 o successiva.

Nota

Per concedere all'account l'accesso alle risorse, in Azure Event Hubs, assegna il ruolo Azure Event Hubs Data Receiver e Azure Event Hubs Data Sender all'account Microsoft Entra attualmente in uso. Quindi, nell'account Azure Storage assegnare il ruolo Storage Blob Data Contributor all'account Microsoft Entra in uso. Per altre informazioni sulla concessione dei ruoli di accesso, vedere Assign Azure roles using the Azure portal and Authorize access to Event Hubs resources using Microsoft Entra ID.

Importante

Spring Boot versione 2.5 o successiva è necessario per completare i passaggi di questa esercitazione.

Inviare e ricevere messaggi da Azure Event Hubs

Con un account Azure Storage e un hub eventi Azure, è possibile inviare e ricevere messaggi usando Spring Cloud Azure Hub eventi di Stream Binder.

Per installare il modulo Spring Cloud Azure Stream Binder Event Hubs, aggiungere le dipendenze seguenti al file pom.xml:

L'elenco dei materiali di Spring Cloud Azure (BOM):
```
<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>com.azure.spring</groupId>
      <artifactId>spring-cloud-azure-dependencies</artifactId>
      <version>7.1.0</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>
```
Nota

Se si usa Spring Boot 4.0.x, assicurarsi di impostare la spring-cloud-azure-dependencies versione su 7.1.0.

Se si usa Spring Boot 3.5.x, assicurarsi di impostare la spring-cloud-azure-dependencies versione su 6.1.0.

Se si usa Spring Boot 3.1.x-3.5.x, assicurarsi di impostare la spring-cloud-azure-dependencies versione su 5.25.0.

Se si usa Spring Boot 2.x, assicurarsi di impostare la spring-cloud-azure-dependencies versione su 4.20.0.

Questo Bill of Material (BOM) deve essere configurato nella sezione <dependencyManagement> del file pom.xml. In questo modo, tutte le dipendenze di Spring Cloud Azure usano la stessa versione.

Per ulteriori informazioni sulla versione utilizzata per questo BOM, vedere Quale versione di Spring Cloud Azure dovrei usare.

L'artefatto di associatore di flusso di Spring Cloud Azure Event Hubs.

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

Codice dell'applicazione

Usare la procedura seguente per configurare l'applicazione per produrre e utilizzare messaggi usando Azure Event Hubs.

Configurare le credenziali dell'hub eventi aggiungendo le proprietà seguenti al file 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

La tabella seguente descrive i campi nella configurazione:

Campo	Descrizione
`spring.cloud.azure.eventhubs.namespace`	Specificare lo spazio dei nomi ottenuto nell'hub eventi dal portale di Azure.
`spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name`	Specificare l'account di archiviazione creato in questa esercitazione.
`spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name`	Specifica il contenitore dell'account di archiviazione.
`spring.cloud.stream.bindings.consume-in-0.destination`	Specificare l'hub eventi usato in questa esercitazione.
`spring.cloud.stream.bindings.consume-in-0.group`	Specifica i gruppi di consumatori nell'istanza di Event Hubs.
`spring.cloud.stream.bindings.supply-out-0.destination`	Specificare lo stesso hub eventi usato in questa esercitazione.
`spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode`	Specificare `MANUAL`.
`spring.cloud.function.definition`	Specificare quale bean funzionale associare alle destinazioni esterne esposte dalle connessioni.
`spring.cloud.stream.poller.initial-delay`	Specificare il ritardo iniziale per i trigger periodici. Il valore predefinito è `0`.
`spring.cloud.stream.poller.fixed-delay`	Specificare il ritardo fisso per il poller predefinito in millisecondi. Il valore predefinito è `1000 L`.

Modificare il file della classe di avvio per visualizzare il contenuto seguente.

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

}

Suggerimento

In questa esercitazione non sono presenti operazioni di autenticazione nelle configurazioni o nel codice. Tuttavia, la connessione ai servizi di Azure richiede l'autenticazione. Per completare l'autenticazione, è necessario usare Azure Identity. Spring Cloud Azure usa DefaultAzureCredential, che la libreria di identità Azure fornisce per ottenere le credenziali senza modifiche al codice.

DefaultAzureCredential supporta più metodi di autenticazione e determina il metodo da usare in fase di esecuzione. Questo approccio consente all'app di usare metodi di autenticazione diversi in ambienti diversi (ad esempio ambienti locali e di produzione) senza implementare codice specifico dell'ambiente. Per altre informazioni, vedere DefaultAzureCredential.

Per completare l'autenticazione negli ambienti di sviluppo locali, è possibile usare Azure CLI, Visual Studio Code, PowerShell o altri metodi. Per altre informazioni, vedere autenticazione Azure in ambienti di sviluppo Java. Per completare l'autenticazione negli ambienti di hosting Azure, è consigliabile usare l'identità gestita assegnata dall'utente. Per altre informazioni, vedere Che sono identità gestite per le risorse Azure?

Avviare l’applicazione. I messaggi simili verranno inseriti nel log applicazioni, come illustrato nell'output di esempio seguente:

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

Distribuire su Azure Spring Apps

Ora che l'applicazione Spring Boot è in esecuzione in locale, è possibile spostarla nell'ambiente di produzione. Azure Spring Apps semplifica la distribuzione di applicazioni Spring Boot in Azure senza modifiche al codice. Il servizio gestisce l'infrastruttura delle applicazioni Spring per consentire agli sviluppatori di concentrarsi sul codice. Azure Spring Apps offre la gestione del ciclo di vita usando monitoraggio e diagnostica completi, gestione della configurazione, individuazione dei servizi, integrazione CI/CD, distribuzioni blu-verde e altro ancora. Per distribuire l'applicazione in Azure Spring Apps, vedere Distribuire la prima applicazione in Azure Spring Apps.

Passaggi successivi

Azure per sviluppatori Spring Spring Cloud Azure Stream Binder Event Hubs Samples

Commenti e suggerimenti

Questa pagina è stata utile?

Last updated on 2026-03-11