Supporto di Spring Cloud Azure per Spring Integration

Si applica a: ✅ Version 4.20.0, ✅ Version 5.25.0, ✅ Version 6.5.0, ✅ Version 7.4.0

L'estensione Spring Integration per Azure offre adattatori Spring Integration per i vari servizi forniti dall'Azure SDK per Java. È disponibile il supporto spring integration per questi servizi di Azure: Hub eventi, bus di servizio, coda di archiviazione. Di seguito è riportato un elenco di adattatori supportati:

spring-cloud-azure-starter-integration-eventhubs - per ulteriori informazioni, vedere Spring Integration con Hub eventi di Azure
spring-cloud-azure-starter-integration-servicebus : per altre informazioni, vedere Spring Integration with bus di servizio di Azure
spring-cloud-azure-starter-integration-storage-queue: per altre informazioni, vedere Spring Integration with Archiviazione di Azure Queue

Integrazione di Spring con Hub eventi di Azure

Concetti chiave

Hub eventi di Azure è una piattaforma di streaming di Big Data e un servizio di inserimento di eventi. Può ricevere ed elaborare milioni di eventi al secondo. I dati inviati a un hub eventi possono essere trasformati e archiviati usando qualsiasi provider di analisi in tempo reale o adattatori di batch/archiviazione.

Spring Integration consente la messaggistica leggera all'interno di applicazioni basate su Spring e supporta l'integrazione con sistemi esterni tramite adattatori dichiarativi. Tali adattatori offrono un livello superiore di astrazione rispetto al supporto di Spring per la comunicazione remota, la messaggistica e la pianificazione. Il progetto di estensione Spring Integration per Event Hubs fornisce adattatori di canale in ingresso e in uscita e gateway per Hub eventi di Azure.

Nota

Le API di supporto di RxJava vengono eliminate dalla versione 4.0.0. Per informazioni dettagliate, vedere Javadoc.

Gruppo di consumatori

Event Hubs offre un supporto simile per i gruppi di consumer di Apache Kafka, ma con una logica leggermente diversa. Mentre Kafka archivia tutti gli offset di cui è stato eseguito il commit nel broker, è necessario archiviare manualmente gli offset dei messaggi di Event Hubs in fase di elaborazione. L'SDK di Event Hubs offre la funzionalità di archiviare tali offset in Archiviazione di Azure.

Supporto per il partizionamento

Event Hubs offre un concetto di partizione fisica simile a quello di Kafka. Tuttavia, a differenza del ribilanciamento automatico di Kafka tra consumer e partizioni, Hub eventi offre una sorta di modalità preemptive. L'account di archiviazione funge da lease per determinare quale consumer possiede quale partizione. Quando viene avviato un nuovo consumer, tenterà di sottrarre alcune partizioni dai consumer più carichi per bilanciare il carico di lavoro.

Per specificare la strategia di bilanciamento del carico, gli sviluppatori possono usare EventHubsContainerProperties per la configurazione. Vedere la sezione seguente per un esempio di come configurare EventHubsContainerProperties.

Supporto per consumer batch

Il EventHubsInboundChannelAdapter supporta la modalità di utilizzo batch. Per abilitarla, gli utenti possono specificare la modalità listener come ListenerMode.BATCH durante la costruzione di un'istanza di EventHubsInboundChannelAdapter. Se abilitata, verrà ricevuto e passato al canale downstream un messaggio di di cui il payload è un elenco di eventi in batch. Ogni intestazione di messaggio viene inoltre convertita in un elenco, il cui contenuto corrisponde al valore dell’intestazione associata estratto da ciascun evento. Per gli header comuni relativi all'ID della partizione, al checkpointer e alle proprietà dell'ultimo evento accodato, viene visualizzato un unico valore per l'intero batch se tutti gli eventi condividono lo stesso valore. Per altre informazioni, vedi la sezione Intestazioni dei messaggi di Event Hubs.

Nota

L'intestazione del checkpoint è presente solo quando si usa la modalità checkpoint MANUAL.

Il checkpoint del consumer batch supporta due modalità: BATCH e MANUAL. BATCH è una modalità di checkpoint automatico che esegue il checkpoint dell'intero batch di eventi nel suo insieme, una volta ricevuto. La modalità MANUAL consiste nel creare un checkpoint degli eventi degli utenti. Se utilizzato, il Checkpointer verrà passato nell'intestazione del messaggio e gli utenti potranno usarlo per eseguire il checkpointing.

I criteri di utilizzo batch possono essere specificati dalle proprietà di max-size e max-wait-time, dove max-size è una proprietà necessaria mentre max-wait-time è facoltativo. Per specificare la strategia di utilizzo batch, gli sviluppatori possono usare EventHubsContainerProperties per la configurazione. Vedere la sezione seguente per un esempio di come configurare EventHubsContainerProperties.

Configurazione delle dipendenze

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-integration-eventhubs</artifactId>
</dependency>

Configurazione

Questo strumento di avvio fornisce le tre parti seguenti delle opzioni di configurazione:

Proprietà di configurazione della connessione

Questa sezione contiene le opzioni di configurazione usate per la connessione a Hub eventi di Azure.

Nota

Se si sceglie di usare un'entità di sicurezza per eseguire l'autenticazione e l'autorizzazione con Microsoft Entra ID per l'accesso a una risorsa di Azure, vedere Autorizzare l'accesso con Microsoft Entra ID per assicurarsi che all'entità di sicurezza sia stata concessa l'autorizzazione sufficiente per accedere alla risorsa di Azure.

Proprietà configurabili della connessione di spring-cloud-azure-starter-integration-eventhubs:

Proprietà	Digitare	Descrizione
`spring.cloud.azure.eventhubs.enabled`	booleano	Indica se hub eventi di Azure è abilitato.
`spring.cloud.azure.eventhubs.connection-string`	Stringa	Valore della stringa di connessione dello spazio dei nomi di Event Hubs.
`spring.cloud.azure.eventhubs.namespace`	Stringa	Valore dello spazio dei nomi di Hub di eventi, ovvero il prefisso del nome di dominio completo. Un FQDN deve essere composto da NamespaceName.DomainName
`spring.cloud.azure.eventhubs.domain-name`	Stringa	Valore del nome di dominio di uno spazio dei nomi di Hub eventi di Azure.
`spring.cloud.azure.eventhubs.custom-endpoint-address`	Stringa	Indirizzo dell'endpoint personalizzato.
`spring.cloud.azure.eventhubs.shared-connection`	Booleano	Se EventProcessorClient e EventHubProducerAsyncClient sottostanti usano la stessa connessione. Per impostazione predefinita, viene creata una nuova connessione e viene usata per ogni client dell'hub eventi creato.

Proprietà di configurazione del checkpoint

Questa sezione contiene le opzioni di configurazione per il servizio BLOB di archiviazione, che viene usato per rendere persistenti le informazioni sulla proprietà della partizione e sul checkpoint.

Nota

Dalla versione 4.0.0, quando la proprietà di spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists non è abilitata manualmente, non verrà creato automaticamente alcun contenitore di archiviazione.

Checkpoint delle proprietà configurabili di spring-cloud-azure-starter-integration-eventhubs:

Proprietà	Digitare	Descrizione
`spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists`	Booleano	Indica se consentire la creazione di contenitori se non esiste.
`spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name`	Stringa	Nome dell'account di archiviazione.
`spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key`	Stringa	Chiave di accesso dell’account di archiviazione.
`spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name`	Stringa	Nome del contenitore di archiviazione.

Le opzioni di configurazione comuni dell'SDK di servizio di Azure sono applicabili anche all'archivio checkpoint di Archiviazione BLOB. Le opzioni di configurazione supportate vengono introdotte in configurazione di Spring Cloud Azuree possono essere configurate con il prefisso unificato spring.cloud.azure. o il prefisso di spring.cloud.azure.eventhubs.processor.checkpoint-store..

Proprietà di configurazione del processore dell'hub eventi

Il EventHubsInboundChannelAdapter usa il EventProcessorClient per utilizzare i messaggi di un hub eventi, per configurare le proprietà complessive di un EventProcessorClient, gli sviluppatori possono usare EventHubsContainerProperties per la configurazione. Vedere sezione seguente su come usare EventHubsInboundChannelAdapter.

Utilizzo di base

Inviare messaggi a Hub eventi di Azure

Compilare le opzioni di configurazione delle credenziali.

Per le credenziali come stringa di connessione, configurare le proprietà seguenti nel file application.yml:
```
spring:
  cloud:
    azure:
      eventhubs:
        connection-string: ${AZURE_EVENT_HUBS_CONNECTION_STRING}
        processor:
          checkpoint-store:
            container-name: ${CHECKPOINT-CONTAINER}
            account-name: ${CHECKPOINT-STORAGE-ACCOUNT}
            account-key: ${CHECKPOINT-ACCESS-KEY}
```
Nota

Microsoft consiglia di usare il flusso di autenticazione più sicuro disponibile. Il flusso di autenticazione descritto in questa procedura, ad esempio per database, cache, messaggistica o servizi di intelligenza artificiale, richiede un livello di attendibilità molto elevato nell'applicazione e comporta rischi non presenti in altri flussi. Usare questo flusso solo quando le opzioni più sicure, ad esempio le identità gestite per le connessioni senza password o senza chiave, non sono valide. Per le operazioni del computer locale, preferire le identità utente per le connessioni senza password o senza chiave.

Per le credenziali come identità gestite, configurare le proprietà seguenti nel file application.yml:

spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${AZURE_CLIENT_ID}
      eventhubs:
        namespace: ${AZURE_EVENT_HUBS_NAMESPACE}
        processor:
          checkpoint-store:
            container-name: ${CONTAINER_NAME}
            account-name: ${ACCOUNT_NAME}

Per le credenziali del service principal, configura le proprietà seguenti nel file application.yml:

spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: <tenant>
      eventhubs:
        namespace: ${AZURE_EVENT_HUBS_NAMESPACE}
        processor:
          checkpoint-store:
            container-name: ${CONTAINER_NAME}
            account-name: ${ACCOUNT_NAME}

Nota

I valori consentiti per tenant-id sono: common, organizations, consumerso l'ID tenant. Per altre informazioni su questi valori, vedere la sezione Usato l'endpoint errato (account personali e dell'organizzazione) di Errore AADSTS50020 - L'account utente del provider di identità non esiste nel tenant. Per informazioni sulla conversione dell'app a tenant singolo, vedere Convertire l'app a tenant singolo in multi-tenant in Microsoft Entra ID.

Crea DefaultMessageHandler con il bean EventHubsTemplate per inviare messaggi a Event Hubs.

class Demo {
    private static final String OUTPUT_CHANNEL = "output";
    private static final String EVENTHUB_NAME = "eh1";

    @Bean
    @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
    public MessageHandler messageSender(EventHubsTemplate eventHubsTemplate) {
        DefaultMessageHandler handler = new DefaultMessageHandler(EVENTHUB_NAME, eventHubsTemplate);
        handler.setSendCallback(new ListenableFutureCallback<Void>() {
            @Override
            public void onSuccess(Void result) {
                LOGGER.info("Message was sent successfully.");
            }
            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error("There was an error sending the message.", ex);
            }
        });
        return handler;
    }
}

Creare un'associazione del gateway di messaggi con il gestore di messaggi sopra indicato tramite un canale di messaggi.

class Demo {
    @Autowired
    EventHubOutboundGateway messagingGateway;

    @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
    public interface EventHubOutboundGateway {
        void send(String text);
    }
}

Inviare messaggi usando il gateway.

class Demo {
    public void demo() {
        this.messagingGateway.send(message);
    }
}

Ricevere messaggi da Hub eventi di Azure

Compilare le opzioni di configurazione delle credenziali.

Creare un bean del canale di messaggi come canale di ingresso.

@Configuration
class Demo {
    @Bean
    public MessageChannel input() {
        return new DirectChannel();
    }
}

Crea EventHubsInboundChannelAdapter con il bean EventHubsMessageListenerContainer per ricevere messaggi da Event Hubs.

@Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";
    private static final String EVENTHUB_NAME = "eh1";
    private static final String CONSUMER_GROUP = "$Default";

    @Bean
    public EventHubsInboundChannelAdapter messageChannelAdapter(
            @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
            EventHubsMessageListenerContainer listenerContainer) {
        EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(listenerContainer);
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }

    @Bean
    public EventHubsMessageListenerContainer messageListenerContainer(EventHubsProcessorFactory processorFactory) {
        EventHubsContainerProperties containerProperties = new EventHubsContainerProperties();
        containerProperties.setEventHubName(EVENTHUB_NAME);
        containerProperties.setConsumerGroup(CONSUMER_GROUP);
        containerProperties.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));
        return new EventHubsMessageListenerContainer(processorFactory, containerProperties);
    }
}

Creare un binding del ricevitore di messaggi con EventHubsInboundChannelAdapter tramite il canale di messaggio creato in precedenza.

class Demo {
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
        String message = new String(payload);
        LOGGER.info("New message received: '{}'", message);
        checkpointer.success()
                .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message))
                .doOnError(e -> LOGGER.error("Error found", e))
                .block();
    }
}

Configurare EventHubsMessageConverter per personalizzare objectMapper

EventHubsMessageConverter viene creato come bean configurabile per consentire agli utenti di personalizzare ObjectMapper.

Supporto per consumer batch

Il consumo in batch dei messaggi da Event Hubs è simile all'esempio precedente; inoltre, gli utenti devono impostare per EventHubsInboundChannelAdapter le opzioni di configurazione relative al consumo in batch.

Quando si crea EventHubsInboundChannelAdapter, la modalità listener deve essere impostata come BATCH. Quando si crea un bean EventHubsMessageListenerContainer, impostare la modalità di checkpoint su MANUAL o BATCH, e le opzioni batch possono essere configurate secondo necessità.

@Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";
    private static final String EVENTHUB_NAME = "eh1";
    private static final String CONSUMER_GROUP = "$Default";

    @Bean
    public EventHubsInboundChannelAdapter messageChannelAdapter(
            @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
            EventHubsMessageListenerContainer listenerContainer) {
        EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(listenerContainer, ListenerMode.BATCH);
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }

    @Bean
    public EventHubsMessageListenerContainer messageListenerContainer(EventHubsProcessorFactory processorFactory) {
        EventHubsContainerProperties containerProperties = new EventHubsContainerProperties();
        containerProperties.setEventHubName(EVENTHUB_NAME);
        containerProperties.setConsumerGroup(CONSUMER_GROUP);
        containerProperties.getBatch().setMaxSize(100);
        containerProperties.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));
        return new EventHubsMessageListenerContainer(processorFactory, containerProperties);
    }
}

Intestazioni dei messaggi di Event Hubs

La tabella seguente mostra come le proprietà dei messaggi di Event Hubs corrispondono alle intestazioni dei messaggi Spring. Per Hub eventi di Azure, il messaggio viene chiamato come event.

Mappatura tra le proprietà di messaggi/eventi di Event Hubs e le intestazioni dei messaggi Spring in modalità listener di record:

Proprietà degli eventi di Event Hubs	Costanti dell'intestazione dei messaggi Spring	Digitare	Descrizione
Ora di accodamento	EventHubsHeaders#ENQUEUED_TIME	Istante	L'istante, in UTC, in cui l'evento è stato accodato nella partizione di Event Hub.
Offset	EventHubsHeaders#OFFSET	Lungo	L'offset dell'evento quando è stato ricevuto dalla partizione di Event Hubs associata.
Chiave di partizione	AzureHeaders#PARTITION_KEY	Stringa	Chiave di hashing della partizione se è stata impostata durante la pubblicazione originale dell'evento.
ID della partizione	AzureHeaders#RAW_PARTITION_ID	Stringa	ID di partizione dell'hub eventi.
Numero di sequenza	EventHubsHeaders#SEQUENCE_NUMBER	Lungo	Numero di sequenza assegnato all'evento al momento dell'accodamento nella partizione di Event Hub associata.
Proprietà dell'ultimo evento accodato	EventHubsHeaders#LAST_ENQUEUED_EVENT_PROPERTIES	Proprietà dell'ultimo evento accodato	Le proprietà dell'ultimo evento accodato in questa partizione.
NA	AzureHeaders#CHECKPOINTER	Posto di blocco	Intestazione per il checkpoint del messaggio specifico.

Gli utenti possono analizzare le intestazioni del messaggio per le informazioni correlate di ogni evento. Per impostare un'intestazione di messaggio per l'evento, tutte le intestazioni personalizzate verranno inserite come proprietà dell'applicazione di un evento, in cui l'intestazione viene impostata come chiave della proprietà. Quando gli eventi vengono ricevuti da Hub eventi, tutte le proprietà dell'applicazione verranno convertite nell'intestazione del messaggio.

Nota

Le intestazioni del messaggio relative a chiave di partizione, ora di accodamento, offset e numero di sequenza non possono essere impostate manualmente.

Quando la modalità consumer batch è abilitata, di seguito sono elencate le intestazioni specifiche dei messaggi raggruppati in batch, che contengono un elenco di valori per ciascun singolo evento di Event Hubs.

Mappatura tra le proprietà dei messaggi/eventi di Event Hubs e le intestazioni dei messaggi Spring in modalità listener batch:

Proprietà degli eventi di Event Hubs	Costanti dell'intestazione del messaggio di Spring Batch	Digitare	Descrizione
Ora di accodamento	EventHubsHeaders#ENQUEUED_TIME	Elenco di Instant	Elenco degli istanti, in UTC, in cui ogni evento è stato accodato nella partizione di Event Hub.
Offset	EventHubsHeaders#OFFSET	Elenco di long	Elenco degli offset di ogni evento al momento della ricezione dalla partizione dell'hub eventi associato.
Chiave di partizione	AzureHeaders#PARTITION_KEY	Elenco di stringhe	Elenco della chiave di hashing della partizione se è stata impostata durante la pubblicazione originale di ogni evento.
Numero di sequenza	EventHubsHeaders#SEQUENCE_NUMBER	Elenco di long	Elenco del numero di sequenza assegnato a ogni evento quando è stato accodato nella partizione di Hub eventi associata.
Proprietà di sistema	EventHubsHeaders#BATCH_CONVERTED_SYSTEM_PROPERTIES	Elenco di mappe	Elenco delle proprietà di sistema di ogni evento.
Proprietà dell'applicazione	EventHubsHeaders#BATCH_CONVERTED_APPLICATION_PROPERTIES	Elenco di mappe	Elenco delle proprietà applicative per ciascun evento, in cui sono riportati tutti gli header dei messaggi personalizzati o le proprietà evento personalizzate.

Nota

Quando si pubblicano i messaggi, tutte le intestazioni batch sopra indicate verranno rimosse dai messaggi, se presenti.

Campioni

Per altre informazioni, vedere il azure-spring-boot-samples repository in GitHub.

Integrazione spring con il bus di servizio di Azure

Concetti chiave

Spring Integration consente la messaggistica leggera all'interno di applicazioni basate su Spring e supporta l'integrazione con sistemi esterni tramite adattatori dichiarativi.

Il progetto di estensione Spring Integration per bus di servizio di Azure fornisce adattatori di canale in entrata e in uscita per bus di servizio di Azure.

Nota

Le API di supporto CompletableFuture sono state deprecate a partire dalla versione 2.10.0 e sono state sostituite da Reactor Core a partire dalla versione 4.0.0. Per informazioni dettagliate, vedere Javadoc.

Configurazione delle dipendenze

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-integration-servicebus</artifactId>
</dependency>

Configurazione

Questo strumento di avvio fornisce le due parti seguenti delle opzioni di configurazione:

Proprietà di configurazione della connessione

Questa sezione contiene le opzioni di configurazione usate per la connessione al bus di servizio di Azure.

Nota

Proprietà configurabili della connessione di spring-cloud-azure-starter-integration-servicebus:

Proprietà	Digitare	Descrizione
`spring.cloud.azure.servicebus.enabled`	booleano	Indica se un bus di servizio di Azure è abilitato.
`spring.cloud.azure.servicebus.connection-string`	Stringa	Valore della stringa di connessione del namespace di bus di servizio.
`spring.cloud.azure.servicebus.custom-endpoint-address`	Stringa	Indirizzo dell'endpoint personalizzato da usare per la connessione al bus di servizio.
`spring.cloud.azure.servicebus.namespace`	Stringa	Valore dello spazio dei nomi di bus di servizio, che corrisponde al prefisso dell'FQDN. Un FQDN deve essere composto da NamespaceName.DomainName
`spring.cloud.azure.servicebus.domain-name`	Stringa	Nome di dominio di un valore dello spazio dei nomi di bus di servizio di Azure.

Proprietà di configurazione del processore del bus di servizio

Il ServiceBusInboundChannelAdapter usa il ServiceBusProcessorClient per utilizzare i messaggi, per configurare le proprietà complessive di un ServiceBusProcessorClient, gli sviluppatori possono usare ServiceBusContainerProperties per la configurazione. Vedere sezione seguente su come usare ServiceBusInboundChannelAdapter.

Utilizzo di base

Inviare messaggi al bus di servizio di Azure

Compilare le opzioni di configurazione delle credenziali.
- Per le credenziali come stringa di connessione, configurare le proprietà seguenti nel file application.yml:
```
spring:
  cloud:
    azure:
      servicebus:
        connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}
```
  Nota
  
  Microsoft consiglia di usare il flusso di autenticazione più sicuro disponibile. Il flusso di autenticazione descritto in questa procedura, ad esempio per database, cache, messaggistica o servizi di intelligenza artificiale, richiede un livello di attendibilità molto elevato nell'applicazione e comporta rischi non presenti in altri flussi. Usare questo flusso solo quando le opzioni più sicure, ad esempio le identità gestite per le connessioni senza password o senza chiave, non sono valide. Per le operazioni del computer locale, preferire le identità utente per le connessioni senza password o senza chiave.
- Per le credenziali come identità gestite, configurare le proprietà seguenti nel file application.yml:
```
spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${AZURE_CLIENT_ID}
      profile:
        tenant-id: <tenant>
      servicebus:
        namespace: ${AZURE_SERVICE_BUS_NAMESPACE}
```

Nota

Per le credenziali del service principal, configura le proprietà seguenti nel file application.yml:

spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: <tenant>
      servicebus:
        namespace: ${AZURE_SERVICE_BUS_NAMESPACE}

Nota

Crea DefaultMessageHandler con il bean ServiceBusTemplate per inviare messaggi a bus di servizio, imposta il tipo di entità per il ServiceBusTemplate. Questo esempio usa una coda di bus di servizio come esempio.

class Demo {
    private static final String OUTPUT_CHANNEL = "queue.output";

    @Bean
    @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
    public MessageHandler queueMessageSender(ServiceBusTemplate serviceBusTemplate) {
        serviceBusTemplate.setDefaultEntityType(ServiceBusEntityType.QUEUE);
        DefaultMessageHandler handler = new DefaultMessageHandler(QUEUE_NAME, serviceBusTemplate);
        handler.setSendCallback(new ListenableFutureCallback<Void>() {
            @Override
            public void onSuccess(Void result) {
                LOGGER.info("Message was sent successfully.");
            }

            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error("There was an error sending the message.", ex);
            }
        });

        return handler;
    }
}

Creare un'associazione del gateway di messaggi con il gestore di messaggi sopra indicato tramite un canale di messaggi.

class Demo {
    @Autowired
    QueueOutboundGateway messagingGateway;

    @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
    public interface QueueOutboundGateway {
        void send(String text);
    }
}

Inviare messaggi usando il gateway.

class Demo {
    public void demo() {
        this.messagingGateway.send(message);
    }
}

Ricevere messaggi dal bus di servizio di Azure

Compilare le opzioni di configurazione delle credenziali.

Creare un bean di canale di messaggi da usare come canale di input.

@Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";

    @Bean
    public MessageChannel input() {
        return new DirectChannel();
    }
}

Crea ServiceBusInboundChannelAdapter con il bean ServiceBusMessageListenerContainer per ricevere messaggi da bus di servizio. Questo esempio usa una coda di bus di servizio come esempio.

@Configuration
class Demo {
    private static final String QUEUE_NAME = "queue1";

    @Bean
    public ServiceBusMessageListenerContainer messageListenerContainer(ServiceBusProcessorFactory processorFactory) {
        ServiceBusContainerProperties containerProperties = new ServiceBusContainerProperties();
        containerProperties.setEntityName(QUEUE_NAME);
        containerProperties.setAutoComplete(false);
        return new ServiceBusMessageListenerContainer(processorFactory, containerProperties);
    }

    @Bean
    public ServiceBusInboundChannelAdapter queueMessageChannelAdapter(
        @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
        ServiceBusMessageListenerContainer listenerContainer) {
        ServiceBusInboundChannelAdapter adapter = new ServiceBusInboundChannelAdapter(listenerContainer);
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }
}

Creare un'associazione di ricevitore di messaggi con ServiceBusInboundChannelAdapter tramite il canale di messaggio creato in precedenza.

class Demo {
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
        String message = new String(payload);
        LOGGER.info("New message received: '{}'", message);
        checkpointer.success()
                .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message))
                .doOnError(e -> LOGGER.error("Error found", e))
                .block();
    }
}

Configurare ServiceBusMessageConverter per personalizzare objectMapper

ServiceBusMessageConverter viene creato come bean configurabile per consentire agli utenti di personalizzare ObjectMapper.

Intestazioni dei messaggi di bus di servizio

Per alcune intestazioni del bus di servizio di cui è possibile eseguire il mapping a più costanti di intestazione Spring, viene elencata la priorità di intestazioni Spring diverse.

Mappatura tra le intestazioni di bus di servizio e le intestazioni di Spring:

Intestazioni e proprietà dei messaggi di bus di servizio	Costanti delle intestazioni dei messaggi di Spring	Digitare	Configurabile	Descrizione
Tipo di contenuto	`MessageHeaders#CONTENT_TYPE`	Stringa	Sì	Il descrittore Content-Type RFC2045 del messaggio.
ID di correlazione	`ServiceBusMessageHeaders#CORRELATION_ID`	Stringa	Sì	ID di correlazione del messaggio
ID messaggio	`ServiceBusMessageHeaders#MESSAGE_ID`	Stringa	Sì	L'ID del messaggio; questa intestazione ha una priorità superiore a `MessageHeaders#ID`.
ID messaggio	`MessageHeaders#ID`	UUID (Identificatore Unico Universale)	Sì	ID messaggio del messaggio, questa intestazione ha una priorità inferiore a `ServiceBusMessageHeaders#MESSAGE_ID`.
Chiave di partizione	`ServiceBusMessageHeaders#PARTITION_KEY`	Stringa	Sì	Chiave di partizione per l'invio del messaggio a un'entità partizionata.
Rispondi a	`MessageHeaders#REPLY_CHANNEL`	Stringa	Sì	Indirizzo di un'entità a cui inviare risposte.
Rispondi all'ID della sessione	`ServiceBusMessageHeaders#REPLY_TO_SESSION_ID`	Stringa	Sì	Il valore della proprietà ReplyToGroupId del messaggio.
Ora di accodamento pianificata utc	`ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME`	OffsetDateTime	Sì	Data e ora in cui il messaggio deve essere messo in coda in bus di servizio; questa intestazione ha una priorità superiore a `AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE`.
Ora di accodamento pianificata utc	`AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE`	Numero intero	Sì	La data e l'ora in cui il messaggio deve essere messo in coda in bus di servizio; questa intestazione ha una priorità inferiore a `ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME`.
ID della sessione	`ServiceBusMessageHeaders#SESSION_ID`	Stringa	Sì	L'identificatore di sessione per un'entità consapevole della sessione.
Tempo di vita (TTL)	`ServiceBusMessageHeaders#TIME_TO_LIVE`	Durata	Sì	Durata dell'intervallo di tempo prima della scadenza del messaggio.
A	`ServiceBusMessageHeaders#TO`	Stringa	Sì	Indirizzo "a" del messaggio, riservato per un uso futuro negli scenari di routing e attualmente ignorato dal broker stesso.
Oggetto	`ServiceBusMessageHeaders#SUBJECT`	Stringa	Sì	Oggetto del messaggio.
Descrizione dell'errore del messaggio non recapitato	`ServiceBusMessageHeaders#DEAD_LETTER_ERROR_DESCRIPTION`	Stringa	NO	Descrizione di un messaggio non recapitato.
Motivo del mancato recapito	`ServiceBusMessageHeaders#DEAD_LETTER_REASON`	Stringa	NO	Il motivo per cui un messaggio è stato inviato alla coda dei messaggi non recapitabili.
Origine dei messaggi non recapitati	`ServiceBusMessageHeaders#DEAD_LETTER_SOURCE`	Stringa	NO	L'entità in cui il messaggio è stato spostato tra i messaggi non recapitabili.
Numero di consegne	`ServiceBusMessageHeaders#DELIVERY_COUNT`	lungo	NO	Numero di volte in cui il messaggio è stato recapitato ai client.
Numero di sequenza accodato	`ServiceBusMessageHeaders#ENQUEUED_SEQUENCE_NUMBER`	lungo	NO	Numero di sequenza accodato assegnato a un messaggio dal bus di servizio.
Ora di accodamento	`ServiceBusMessageHeaders#ENQUEUED_TIME`	OffsetDateTime	NO	Data e ora in cui questo messaggio è stato accodato in bus di servizio.
Scade il	`ServiceBusMessageHeaders#EXPIRES_AT`	OffsetDateTime	NO	Data/ora in cui il messaggio scadrà.
Token di blocco	`ServiceBusMessageHeaders#LOCK_TOKEN`	Stringa	NO	Token di blocco per il messaggio corrente.
Bloccato fino a quando	`ServiceBusMessageHeaders#LOCKED_UNTIL`	OffsetDateTime	NO	Data/ora in cui scade il blocco del messaggio.
Numero di sequenza	`ServiceBusMessageHeaders#SEQUENCE_NUMBER`	lungo	NO	Numero univoco assegnato a un messaggio dal bus di servizio.
Stato	`ServiceBusMessageHeaders#STATE`	ServiceBusMessageState	NO	Stato del messaggio, che può essere Attivo, Posticipato o Pianificato.

Supporto per la chiave di partizione

Questo starter supporta il partizionamento di bus di servizio consentendo di impostare la chiave di partizione e l'ID di sessione nell'intestazione del messaggio. Questa sezione illustra come impostare la chiave di partizione per i messaggi.

Consigliato: usare ServiceBusMessageHeaders.PARTITION_KEY come chiave dell'intestazione.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(ServiceBusMessageHeaders.PARTITION_KEY, "Customize partition key")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Non consigliato ma attualmente supportato: AzureHeaders.PARTITION_KEY come chiave dell'intestazione.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(AzureHeaders.PARTITION_KEY, "Customize partition key")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Nota

Quando sia ServiceBusMessageHeaders.PARTITION_KEY che AzureHeaders.PARTITION_KEY vengono impostati nelle intestazioni del messaggio, è preferibile ServiceBusMessageHeaders.PARTITION_KEY.

Supporto delle sessioni

In questo esempio viene illustrato come impostare manualmente l'ID sessione di un messaggio nell'applicazione.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session ID")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Nota

Quando la ServiceBusMessageHeaders.SESSION_ID viene impostata nelle intestazioni del messaggio e viene impostata anche un'intestazione di ServiceBusMessageHeaders.PARTITION_KEY diversa, il valore dell'ID sessione verrà infine usato per sovrascrivere il valore della chiave di partizione.

Personalizzare le proprietà del client di bus di servizio

Gli sviluppatori possono usare AzureServiceClientBuilderCustomizer per personalizzare le proprietà client del bus di servizio. Nell'esempio seguente viene personalizzata la proprietà sessionIdleTimeout in ServiceBusClientBuilder:

@Bean
public AzureServiceClientBuilderCustomizer<ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder> customizeBuilder() {
    return builder -> builder.sessionIdleTimeout(Duration.ofSeconds(10));
}

Campioni

Per altre informazioni, vedere il azure-spring-boot-samples repository in GitHub.

Integrazione di Spring con Coda di archiviazione di Azure

Concetti chiave

archiviazione code di Azure è un servizio per archiviare grandi quantità di messaggi. È possibile accedere ai messaggi da qualsiasi parte del mondo tramite chiamate autenticate tramite HTTP o HTTPS. Un messaggio in coda può avere dimensioni fino a 64 KB. Una coda può contenere milioni di messaggi, fino al limite di capacità totale di un account di archiviazione. Le code vengono comunemente usate per creare un backlog di lavoro da elaborare in modo asincrono.

Configurazione delle dipendenze

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-integration-storage-queue</artifactId>
</dependency>

Configurazione

Questo strumento di avvio offre le opzioni di configurazione seguenti:

Proprietà di configurazione della connessione

Questa sezione contiene le opzioni di configurazione usate per la connessione alla coda di archiviazione di Azure.

Nota

Proprietà configurabili della connessione di spring-cloud-azure-starter-integration-storage-queue:

Proprietà	Digitare	Descrizione
`spring.cloud.azure.storage.queue.enabled`	booleano	Indica se è abilitata una coda di archiviazione di Azure.
`spring.cloud.azure.storage.queue.connection-string`	Stringa	Valore della stringa di connessione dello spazio dei nomi della coda di archiviazione.
`spring.cloud.azure.storage.queue.accountName`	Stringa	Nome dell'account della coda di archiviazione.
`spring.cloud.azure.storage.queue.accountKey`	Stringa	Chiave dell'account della queue di archiviazione.
`spring.cloud.azure.storage.queue.endpoint`	Stringa	Endpoint del servizio di accodamento archiviazione.
`spring.cloud.azure.storage.queue.sasToken`	Stringa	Credenziale del token SAS
`spring.cloud.azure.storage.queue.serviceVersion`	QueueServiceVersion	QueueServiceVersion usato durante l'esecuzione di richieste API.
`spring.cloud.azure.storage.queue.messageEncoding`	Stringa	Codifica dei messaggi della coda.

Utilizzo di base

Inviare messaggi alla coda di archiviazione di Azure

Compilare le opzioni di configurazione delle credenziali.
- Per le credenziali come stringa di connessione, configurare le proprietà seguenti nel file application.yml:
```
spring:
  cloud:
    azure:
      storage:
        queue:
          connection-string: ${AZURE_STORAGE_QUEUE_CONNECTION_STRING}
```
  Nota
  
  Microsoft consiglia di usare il flusso di autenticazione più sicuro disponibile. Il flusso di autenticazione descritto in questa procedura, ad esempio per database, cache, messaggistica o servizi di intelligenza artificiale, richiede un livello di attendibilità molto elevato nell'applicazione e comporta rischi non presenti in altri flussi. Usare questo flusso solo quando le opzioni più sicure, ad esempio le identità gestite per le connessioni senza password o senza chiave, non sono valide. Per le operazioni del computer locale, preferire le identità utente per le connessioni senza password o senza chiave.
- Per le credenziali come identità gestite, configurare le proprietà seguenti nel file application.yml:
```
spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${AZURE_CLIENT_ID}
      profile:
        tenant-id: <tenant>
      storage:
        queue:
          account-name: ${AZURE_STORAGE_QUEUE_ACCOUNT_NAME}
```

Nota

Per le credenziali del service principal, configura le proprietà seguenti nel file application.yml:

spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: <tenant>
      storage:
        queue:
          account-name: ${AZURE_STORAGE_QUEUE_ACCOUNT_NAME}

Nota

Crea DefaultMessageHandler con il bean StorageQueueTemplate per inviare messaggi a Storage Queue.

class Demo {
    private static final String STORAGE_QUEUE_NAME = "example";
    private static final String OUTPUT_CHANNEL = "output";

    @Bean
    @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
    public MessageHandler messageSender(StorageQueueTemplate storageQueueTemplate) {
        DefaultMessageHandler handler = new DefaultMessageHandler(STORAGE_QUEUE_NAME, storageQueueTemplate);
        handler.setSendCallback(new ListenableFutureCallback<Void>() {
            @Override
            public void onSuccess(Void result) {
                LOGGER.info("Message was sent successfully.");
            }

            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error("There was an error sending the message.", ex);
            }
        });
        return handler;
    }
}

Creare un binding del gateway di messaggi con il gestore dei messaggi sopra indicato tramite un canale di messaggi.

class Demo {
    @Autowired
    StorageQueueOutboundGateway storageQueueOutboundGateway;

    @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
    public interface StorageQueueOutboundGateway {
        void send(String text);
    }
}

Inviare messaggi usando il gateway.

class Demo {
    public void demo() {
        this.storageQueueOutboundGateway.send(message);
    }
}

Ricevere messaggi dalla coda di archiviazione di Azure

Compilare le opzioni di configurazione delle credenziali.

Creare un bean del canale di messaggi come canale di ingresso.

class Demo {
    private static final String INPUT_CHANNEL = "input";

    @Bean
    public MessageChannel input() {
        return new DirectChannel();
    }
}

Crea StorageQueueMessageSource con il bean StorageQueueTemplate per ricevere i messaggi da Storage Queue.

class Demo {
    private static final String STORAGE_QUEUE_NAME = "example";

    @Bean
    @InboundChannelAdapter(channel = INPUT_CHANNEL, poller = @Poller(fixedDelay = "1000"))
    public StorageQueueMessageSource storageQueueMessageSource(StorageQueueTemplate storageQueueTemplate) {
        return new StorageQueueMessageSource(STORAGE_QUEUE_NAME, storageQueueTemplate);
    }
}

Creare un binding del ricevitore di messaggi con StorageQueueMessageSource creato nell'ultimo passaggio tramite il canale di messaggio creato in precedenza.

class Demo {
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
        String message = new String(payload);
        LOGGER.info("New message received: '{}'", message);
        checkpointer.success()
            .doOnError(Throwable::printStackTrace)
            .doOnSuccess(t -> LOGGER.info("Message '{}' successfully checkpointed", message))
            .block();
    }
}

Campioni

Per altre informazioni, vedere il azure-spring-boot-samples repository in GitHub.

Commenti e suggerimenti

Questa pagina è stata utile?

Ultimo aggiornamento: 2026-09-21

Supporto di Spring Cloud Azure per Spring Integration

Integrazione di Spring con Hub eventi di Azure

Concetti chiave

Gruppo di consumatori

Supporto per il partizionamento

Supporto per consumer batch

Configurazione delle dipendenze

Configurazione

Proprietà di configurazione della connessione

Proprietà di configurazione del checkpoint

Proprietà di configurazione del processore dell'hub eventi

Utilizzo di base

Inviare messaggi a Hub eventi di Azure

Ricevere messaggi da Hub eventi di Azure

Configurare EventHubsMessageConverter per personalizzare objectMapper

Supporto per consumer batch

Intestazioni dei messaggi di Event Hubs

Campioni

Integrazione spring con il bus di servizio di Azure

Concetti chiave

Configurazione delle dipendenze

Configurazione

Proprietà di configurazione della connessione

Proprietà di configurazione del processore del bus di servizio

Utilizzo di base

Inviare messaggi al bus di servizio di Azure

Ricevere messaggi dal bus di servizio di Azure

Configurare ServiceBusMessageConverter per personalizzare objectMapper

Intestazioni dei messaggi di bus di servizio

Supporto per la chiave di partizione

Supporto delle sessioni

Personalizzare le proprietà del client di bus di servizio

Campioni

Integrazione di Spring con Coda di archiviazione di Azure

Concetti chiave

Configurazione delle dipendenze

Configurazione

Proprietà di configurazione della connessione

Utilizzo di base

Inviare messaggi alla coda di archiviazione di Azure

Ricevere messaggi dalla coda di archiviazione di Azure

Campioni

Commenti e suggerimenti

Risorse aggiuntive