Spring Cloud Stream con bus di servizio di Azure

Questo articolo illustra come usare Spring Cloud Stream Binder per inviare e ricevere messaggi da bus di servizio queues e topics.

Azure fornisce una piattaforma di messaggistica asincrona denominata bus di servizio di Azure ("bus di servizio") basata sullo standard Advanced Message Queueing Protocol 1.0 ("AMQP 1.0"). bus di servizio possono essere usati in tutta la gamma di piattaforme di Azure supportate.

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

• Coda o argomento per bus di servizio di Azure. Se non ne hai uno, crea una coda di bus di servizio o crea un argomento bus di servizio.

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

Inviare e ricevere messaggi da bus di servizio di Azure

Con una coda o un argomento per bus di servizio di Azure, è possibile inviare e ricevere messaggi usando Spring Cloud Azure Stream Binder bus di servizio.

Per installare il modulo Spring Cloud Azure Stream Binder-bus di servizio, aggiungere le seguenti dipendenze al file pom.xml:

• Spring Cloud Azure Bill of Materials (Elenco dei Materiali, BOM):

  <dependencyManagement>
    <dependencies>
      <dependency>
        <groupId>com.azure.spring</groupId>
        <artifactId>spring-cloud-azure-dependencies</artifactId>
        <version>7.2.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.2.0.

  Se si usa Spring Boot 3.5.x, assicurarsi di impostare la spring-cloud-azure-dependencies versione su 6.2.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, consultare Quale versione di Spring Cloud Azure dovrei usare.

• Artefatto bus di servizio Spring Cloud Azure Stream Binder:

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

Codice dell'applicazione

Usare la procedura seguente per configurare l'applicazione in modo da usare una coda o un argomento bus di servizio per inviare e ricevere messaggi.

1. Configurare le credenziali di bus di servizio nel file di configurazione application.properties.

   • Usare una coda di bus di servizio
   • Usare un argomento bus di servizio

    spring.cloud.azure.servicebus.namespace=${AZURE_SERVICEBUS_NAMESPACE}
    spring.cloud.stream.bindings.consume-in-0.destination=${AZURE_SERVICEBUS_QUEUE_NAME}
    spring.cloud.stream.bindings.supply-out-0.destination=${AZURE_SERVICEBUS_QUEUE_NAME}
    spring.cloud.stream.servicebus.bindings.consume-in-0.consumer.auto-complete=false
    spring.cloud.stream.servicebus.bindings.supply-out-0.producer.entity-type=queue
    spring.cloud.function.definition=consume;supply;
    spring.cloud.stream.poller.fixed-delay=60000 
    spring.cloud.stream.poller.initial-delay=0
   

   La tabella seguente descrive i campi nella configurazione:

   Campo	Descrizione
   spring.cloud.azure.servicebus.namespace	Specifica lo spazio dei nomi ottenuto nel tuo bus di servizio dal portale di Azure.
   spring.cloud.stream.bindings.consume-in-0.destination	Specificare la coda di bus di servizio o l'argomento di bus di servizio utilizzato in questa esercitazione.
   spring.cloud.stream.bindings.supply-out-0.destination	Specificare lo stesso valore usato per la destinazione input.
   spring.cloud.stream.servicebus.bindings.consume-in-0.consumer.auto-complete	Specificare se risolvere automaticamente i messaggi. Se impostato come false, verrà aggiunta un'intestazione di messaggio di Checkpointer per permettere agli sviluppatori di gestire manualmente i messaggi.
   spring.cloud.stream.servicebus.bindings.supply-out-0.producer.entity-type	Specificare il tipo di entità per l'associazione di output, può essere queue o topic.
   spring.cloud.function.definition	Specificare il bean funzionale da associare alle destinazioni esterne esposte dai binding.
   spring.cloud.stream.poller.fixed-delay	Definire il ritardo fisso per il poller predefinito in millisecondi. Il valore predefinito è 1000 L. Il valore consigliato è 60000.
   spring.cloud.stream.poller.initial-delay	Specificare il ritardo iniziale per i trigger periodici. Il valore predefinito è 0.

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

   import com.azure.spring.messaging.checkpoint.Checkpointer;
   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 ServiceBusQueueBinderApplication implements CommandLineRunner {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueBinderApplication.class);
       private static final Sinks.Many<Message<String>> many = Sinks.many().unicast().onBackpressureBuffer();
   
       public static void main(String[] args) {
           SpringApplication.run(ServiceBusQueueBinderApplication.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: '{}'", message.getPayload());
               checkpointer.success()
                           .doOnSuccess(s->LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                           .doOnError(e->LOGGER.error("Error found", e))
                           .block();
           };
       }
   
       @Override
       public void run(String... args) {
           LOGGER.info("Going to add message {} to Sinks.Many.", "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 interfaccia della riga di comando di Azure, 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?

3. Avviare l’applicazione. I messaggi come l'esempio seguente verranno inseriti nel log applicazioni:

   New message received: 'Hello World'
   Message 'Hello World' successfully checkpointed
   

Passaggi successivi