Nota
L'accesso a questa pagina richiede l'autorizzazione. È possibile provare ad accedere o modificare le directory.
L'accesso a questa pagina richiede l'autorizzazione. È possibile provare a modificare le directory.
Spring Cloud Stream è un framework per la creazione di microservizi altamente scalabili basati su eventi connessi con sistemi di messaggistica condivisa.
Il framework fornisce un modello di programmazione flessibile basato su idiomi e procedure consigliate Spring già stabilite e familiari. Queste procedure consigliate includono il supporto per una semantica pub/sub persistente, gruppi di consumer e partizioni con stato.
Le implementazioni attuali del binder includono:
-
spring-cloud-azure-stream-binder-eventhubs: per altre informazioni, vedere Spring Cloud Stream Binder for Hub eventi di Azure -
spring-cloud-azure-stream-binder-servicebus: per altre informazioni, vedere Spring Cloud Stream Binder for bus di servizio di Azure
Spring Cloud Stream Binder per Hub eventi di Azure
Concetti chiave
Il binder di Spring Cloud Stream per Hub eventi di Azure fornisce l'implementazione del binding per il framework Spring Cloud Stream. Questa implementazione usa Spring Integration Event Hubs Channel Adapters alla sua base. Dal punto di vista della progettazione, Hub eventi è simile a Kafka. È anche possibile accedere a Hub eventi tramite l'API Kafka. Se il progetto ha una dipendenza stretta dall'API Kafka, è possibile provare Hub eventi con l'esempio di API Kafka
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 di Kafka, che esegue il ribilanciamento automatico tra consumer e partizioni, Event Hubs offre una sorta di modalità preventiva. L'account di archiviazione funge da meccanismo di lease per determinare quale consumer è proprietario di quale partizione. Quando viene avviato un nuovo consumer, questo tenta di sottrarre alcune partizioni ai consumer con il carico maggiore per bilanciare il carico di lavoro.
Per specificare la strategia di bilanciamento del carico, vengono fornite proprietà di spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.load-balancing.*. Per ulteriori informazioni, vedere la sezione Proprietà Consumer.
Supporto per consumer batch
Il binder Spring Cloud Azure Stream Event Hubs supporta la funzionalità Batch Consumer di Spring Cloud Stream.
Per usare la modalità batch-consumer, impostare la proprietà spring.cloud.stream.bindings.<binding-name>.consumer.batch-mode su true. Se abilitata, viene ricevuto un messaggio con un payload di un elenco di eventi in batch e passato alla funzione Consumer. Anche ogni intestazione del messaggio viene convertita in un elenco, il cui contenuto corrisponde al valore dell’intestazione associata ottenuto dall’analisi di ciascun evento. Le intestazioni condivise dell'ID della partizione, del checkpointer e delle proprietà dell'ultimo accodamento vengono presentate come un unico valore perché l'intero batch di eventi condivide lo stesso valore. Per altre informazioni, vedere la sezione intestazioni dei messaggi di Hub eventi di supporto di Spring Cloud Azure per Spring Integration.
Nota
L'intestazione del checkpoint è presente solo quando viene utilizzata la modalità checkpoint MANUAL.
Il checkpoint del consumer batch supporta due modalità: BATCH e MANUAL.
BATCH è una modalità di checkpoint automatico che crea un checkpoint dell'intero batch di eventi non appena il binder li riceve. La modalità MANUAL consiste nel creare un checkpoint degli eventi degli utenti. Se usato, il Checkpointer viene passato nell'intestazione del messaggio e gli utenti possono usarlo per eseguire il checkpoint.
È possibile specificare le dimensioni del batch impostando le proprietà max-size e max-wait-time con prefisso spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.batch.. La proprietà max-size è necessaria e la proprietà max-wait-time è facoltativa. Per ulteriori informazioni, vedere la sezione Proprietà consumer.
Configurazione delle dipendenze
<dependency>
<groupId>com.azure.spring</groupId>
<artifactId>spring-cloud-azure-stream-binder-eventhubs</artifactId>
</dependency>
In alternativa, è anche possibile usare Spring Cloud Azure Stream Event Hubs Starter, come illustrato nell'esempio seguente per Maven:
<dependency>
<groupId>com.azure.spring</groupId>
<artifactId>spring-cloud-azure-starter-stream-eventhubs</artifactId>
</dependency>
Configurazione
Il binder 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-stream-binder-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. |
Suggerimento
Anche il binder Spring Cloud Azure Stream per Event Hubs supporta la configurazione delle comuni opzioni dell'SDK dei servizi Azure. 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..
Il binder supporta anche Spring Could Azure Resource Manager per impostazione predefinita. Per informazioni su come recuperare la stringa di connessione con principal di sicurezza a cui non sono stati concessi i ruoli correlati a Data, vedere la sezione Utilizzo di base di Spring Cloud Azure Resource Manager.
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, nessun contenitore di archiviazione verrà creato automaticamente con il nome da spring.cloud.stream.bindings.binding-name.destination.
Checkpoint delle proprietà configurabili di spring-cloud-azure-stream-binder-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. |
Suggerimento
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 binding di Hub eventi di Azure
Le opzioni seguenti sono suddivise in quattro sezioni: Proprietà del consumatore, Configurazioni avanzate del consumatore, Proprietà del produttore e Configurazioni avanzate del produttore.
Proprietà del consumatore
Queste proprietà vengono esposte tramite EventHubsConsumerProperties.
Nota
Per evitare ripetizioni, a partire dalle versioni 4.17.0 e 5.11.0, Spring Cloud Azure Stream Binder Event Hubs supporta l'impostazione di valori per tutti i canali, nel formato spring.cloud.stream.eventhubs.default.consumer.<property>=<value>.
Proprietà di spring-cloud-azure-stream-binder-eventhubs configurabili dall'utente:
| Proprietà | Digitare | Descrizione |
|---|---|---|
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.checkpoint.mode |
Modalità punto di controllo | Modalità checkpoint utilizzata quando il consumer decide come effettuare il checkpoint del messaggio |
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.checkpoint.count |
Numero intero | Decide la quantità di messaggio per ogni partizione per eseguire un checkpoint. Avrà effetto solo quando viene usata PARTITION_COUNT modalità checkpoint. |
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.checkpoint.interval |
Durata | Decide l'intervallo di tempo per eseguire un checkpoint. Avrà effetto solo quando viene usata TIME modalità checkpoint. |
spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.batch.max-size |
Numero intero | Numero massimo di eventi in un batch. Obbligatorio per la modalità batch-consumer. |
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.batch.max-wait-time |
Durata | Durata massima del tempo per l'utilizzo in batch. Avrà effetto solo quando la modalità batch-consumer è abilitata ed è facoltativa. |
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.load-balancing.update-interval |
Durata | Durata dell'intervallo per l'aggiornamento. |
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.load-balancing.strategy |
Strategia di bilanciamento del carico | Strategia di bilanciamento del carico. |
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.load-balancing.partition-ownership-expiration-interval |
Durata | L'intervallo di tempo dopo il quale scade la proprietà della partizione. |
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.track-last-enqueued-event-properties |
Booleano | Indica se il processore di eventi deve richiedere informazioni sull'ultimo evento accodato nella partizione associata e tenere traccia di tali informazioni man mano che gli eventi vengono ricevuti. |
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.prefetch-count |
Numero intero | Il conteggio usato dal consumer per controllare il numero di eventi che il consumer dell'hub eventi riceverà e accoderà attivamente in locale. |
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.initial-partition-event-position |
Mappa con l'ID della partizione come chiave e i valori di StartPositionProperties |
Mappa contenente la posizione dell'evento da usare per ogni partizione se non esiste un checkpoint per la partizione nell'archivio checkpoint. Questa mappa è indicizzata in base all'ID della partizione. |
Nota
La configurazione initial-partition-event-position accetta un map per specificare la posizione iniziale per ogni hub eventi. Pertanto, la chiave è l'ID della partizione e il valore è StartPositionProperties, che include le proprietà di offset, numero di sequenza, data e ora di accodamento e se è inclusivo. Ad esempio, è possibile impostarlo come
spring:
cloud:
stream:
eventhubs:
bindings:
<binding-name>:
consumer:
initial-partition-event-position:
0:
offset: earliest
1:
sequence-number: 100
2:
enqueued-date-time: 2022-01-12T13:32:47.650005Z
4:
inclusive: false
Configurazione avanzata utente
Le configurazioni di connection, checkpoint e common Azure SDK client sopra indicate supportano la personalizzazione per ogni consumer del binder, che è possibile configurare con il prefisso spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer..
Proprietà del produttore
Queste proprietà vengono esposte tramite EventHubsProducerProperties.
Nota
Per evitare ripetizioni, a partire dalle versioni 4.17.0 e 5.11.0, Spring Cloud Azure Stream Binder Event Hubs supporta l'impostazione di valori per tutti i canali, nel formato spring.cloud.stream.eventhubs.default.producer.<property>=<value>.
Proprietà di spring-cloud-azure-stream-binder-eventhubs configurabili dal produttore:
| Proprietà | Digitare | Descrizione |
|---|---|---|
spring.cloud.stream.eventhubs.bindings.binding-name.producer.sync |
booleano | Flag di attivazione per la sincronizzazione del produttore. Se è vero, il produttore attenderà una risposta dopo un'operazione di invio. |
spring.cloud.stream.eventhubs.bindings.binding-name.producer.send-timeout |
lungo | Quantità di tempo di attesa per una risposta dopo un'operazione di invio. Avrà effetto solo se è abilitato un produttore di sincronizzazione. |
Configurazione avanzata del produttore
La configurazione precedente di connessione e del client comune di Azure SDK supporta la personalizzazione per ogni producer del binder, che è possibile configurare con il prefisso spring.cloud.stream.eventhubs.bindings.<binding-name>.producer..
Utilizzo di base
Invio e ricezione di messaggi da/a Hub eventi
Compilare le opzioni di configurazione con le informazioni sulle credenziali.
Per le credenziali come stringa di connessione, configurare le proprietà seguenti nel file application.yml:
spring: cloud: azure: eventhubs: connection-string: ${EVENTHUB_NAMESPACE_CONNECTION_STRING} processor: checkpoint-store: container-name: ${CHECKPOINT_CONTAINER} account-name: ${CHECKPOINT_STORAGE_ACCOUNT} account-key: ${CHECKPOINT_ACCESS_KEY} function: definition: consume;supply stream: bindings: consume-in-0: destination: ${EVENTHUB_NAME} group: ${CONSUMER_GROUP} supply-out-0: destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE} eventhubs: bindings: consume-in-0: consumer: checkpoint: mode: MANUALNota
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 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: ${EVENTHUB_NAMESPACE} processor: checkpoint-store: container-name: ${CONTAINER_NAME} account-name: ${ACCOUNT_NAME} function: definition: consume;supply stream: bindings: consume-in-0: destination: ${EVENTHUB_NAME} group: ${CONSUMER_GROUP} supply-out-0: destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE} eventhubs: bindings: consume-in-0: consumer: checkpoint: mode: MANUAL
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.
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_MANAGED_IDENTITY_CLIENT_ID} # Only needed when using a user-assigned managed identity eventhubs: namespace: ${EVENTHUB_NAMESPACE} processor: checkpoint-store: container-name: ${CONTAINER_NAME} account-name: ${ACCOUNT_NAME} function: definition: consume;supply stream: bindings: consume-in-0: destination: ${EVENTHUB_NAME} group: ${CONSUMER_GROUP} supply-out-0: destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE} eventhubs: bindings: consume-in-0: consumer: checkpoint: mode: MANUAL
Definire i fornitori e i consumatori.
@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(); }; } @Bean public Supplier<Message<String>> supply() { return () -> { LOGGER.info("Sending message, sequence " + i); return MessageBuilder.withPayload("Hello world, " + i++).build(); }; }
Supporto per il partizionamento
Viene creata una PartitionSupplier con informazioni sulla partizione fornite dall'utente per configurare le informazioni sulla partizione per il messaggio da inviare. Il diagramma di flusso seguente mostra il processo per ottenere diverse priorità per l'ID di partizione e la chiave:
Supporto per consumer batch
Specificare le opzioni di configurazione batch, come illustrato nell'esempio seguente:
spring: cloud: function: definition: consume stream: bindings: consume-in-0: destination: ${AZURE_EVENTHUB_NAME} group: ${AZURE_EVENTHUB_CONSUMER_GROUP} consumer: batch-mode: true eventhubs: bindings: consume-in-0: consumer: batch: max-batch-size: 10 # Required for batch-consumer mode max-wait-time: 1m # Optional, the default value is null checkpoint: mode: BATCH # or MANUAL as neededDefinire i fornitori e i consumatori.
Con la modalità checkpoint impostata su
BATCH, puoi usare il codice seguente per inviare messaggi ed elaborarli in batch.@Bean public Consumer<Message<List<String>>> consume() { return message -> { for (int i = 0; i < message.getPayload().size(); i++) { LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}", message.getPayload().get(i), ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_PARTITION_KEY)).get(i), ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_SEQUENCE_NUMBER)).get(i), ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_OFFSET)).get(i), ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_ENQUEUED_TIME)).get(i)); } }; } @Bean public Supplier<Message<String>> supply() { return () -> { LOGGER.info("Sending message, sequence " + i); return MessageBuilder.withPayload("\"test"+ i++ +"\"").build(); }; }Per la modalità di checkpoint come
MANUAL, è possibile usare il codice seguente per inviare messaggi e utilizzare/checkpoint in batch.@Bean public Consumer<Message<List<String>>> consume() { return message -> { for (int i = 0; i < message.getPayload().size(); i++) { LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}", message.getPayload().get(i), ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_PARTITION_KEY)).get(i), ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_SEQUENCE_NUMBER)).get(i), ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_OFFSET)).get(i), ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_ENQUEUED_TIME)).get(i)); } Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER); checkpointer.success() .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload())) .doOnError(error -> LOGGER.error("Exception found", error)) .block(); }; } @Bean public Supplier<Message<String>> supply() { return () -> { LOGGER.info("Sending message, sequence " + i); return MessageBuilder.withPayload("\"test"+ i++ +"\"").build(); }; }
Nota
Nella modalità di utilizzo batch, il tipo di contenuto predefinito di Spring Cloud Stream Binder è application/json, quindi assicurarsi che il payload del messaggio sia allineato al tipo di contenuto. Ad esempio, quando si usa il tipo di contenuto predefinito di application/json per ricevere messaggi con payload String, il payload deve essere JSON String, racchiuso tra virgolette doppie per il testo originale String. Mentre per il tipo di contenuto text/plain, può essere direttamente un oggetto String. Per altre informazioni, vedere Spring Cloud Stream Content Type Negotiation.
Gestire i messaggi di errore
- Spring Cloud Azure 6.x/7.x
- Spring Cloud Azure 5.x
- Spring Cloud Azure 4.x
Gestire i messaggi di errore di associazione in uscita
Per impostazione predefinita, Spring Integration crea un canale di errore globale denominato
errorChannel. Configurare l'endpoint del messaggio seguente per gestire i messaggi di errore di associazione in uscita.@ServiceActivator(inputChannel = IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) public void handleError(ErrorMessage message) { LOGGER.error("Handling outbound binding error: " + message); }Gestire i messaggi di errore di associazione in ingresso
Spring Cloud Stream Event Hubs Binder supporta una soluzione per gestire gli errori per le associazioni di messaggi in ingresso: gestori degli errori.
Gestore errori:
Spring Cloud Stream espone un meccanismo per fornire un gestore degli errori personalizzato aggiungendo un
Consumerche accetta istanze diErrorMessage. Per altre informazioni, vedere Gestire i messaggi di errore nella documentazione di Spring Cloud Stream.Gestore di errori predefinito per il binding
Configurare un singolo bean
Consumerper gestire tutti i messaggi di errore del binding in ingresso. La seguente funzione predefinita si sottoscrive a ciascun canale di errore dei binding in ingresso:@Bean public Consumer<ErrorMessage> myDefaultHandler() { return message -> { // consume the error message }; }È anche necessario impostare la proprietà
spring.cloud.stream.default.error-handler-definitionsul nome della funzione.Gestore degli errori specifico del binding
Configurare un
Consumerbean per gestire gli specifici messaggi di errore del binding in ingresso. La funzione seguente si sottoscrive al canale di errore specifico del binding in ingresso e ha una priorità più alta rispetto al gestore errori predefinito del binding:@Bean public Consumer<ErrorMessage> myErrorHandler() { return message -> { // consume the error message }; }È anche necessario impostare la proprietà
spring.cloud.stream.bindings.<input-binding-name>.error-handler-definitionsul nome della funzione.
Intestazioni dei messaggi di Event Hubs
Per informazioni sulle intestazioni di base dei messaggi supportate, consultare la sezione intestazioni dei messaggi di Event Hubs di supporto di Spring Cloud Azure per Spring Integration.
Supporto di più binder
È supportata anche la connessione a più namespace di Event Hubs tramite binder multipli. Questo esempio usa una stringa di connessione. Sono supportate anche le credenziali delle entità servizio e delle identità gestite. È possibile impostare le relative proprietà nelle impostazioni dell'ambiente di ciascun binder.
Per usare più binder con Hub eventi, configurare le proprietà seguenti nel file application.yml:
spring: cloud: function: definition: consume1;supply1;consume2;supply2 stream: bindings: consume1-in-0: destination: ${EVENTHUB_NAME_01} group: ${CONSUMER_GROUP_01} supply1-out-0: destination: ${THE_SAME_EVENTHUB_NAME_01_AS_ABOVE} consume2-in-0: binder: eventhub-2 destination: ${EVENTHUB_NAME_02} group: ${CONSUMER_GROUP_02} supply2-out-0: binder: eventhub-2 destination: ${THE_SAME_EVENTHUB_NAME_02_AS_ABOVE} binders: eventhub-1: type: eventhubs default-candidate: true environment: spring: cloud: azure: eventhubs: connection-string: ${EVENTHUB_NAMESPACE_01_CONNECTION_STRING} processor: checkpoint-store: container-name: ${CHECKPOINT_CONTAINER_01} account-name: ${CHECKPOINT_STORAGE_ACCOUNT} account-key: ${CHECKPOINT_ACCESS_KEY} eventhub-2: type: eventhubs default-candidate: false environment: spring: cloud: azure: eventhubs: connection-string: ${EVENTHUB_NAMESPACE_02_CONNECTION_STRING} processor: checkpoint-store: container-name: ${CHECKPOINT_CONTAINER_02} account-name: ${CHECKPOINT_STORAGE_ACCOUNT} account-key: ${CHECKPOINT_ACCESS_KEY} eventhubs: bindings: consume1-in-0: consumer: checkpoint: mode: MANUAL consume2-in-0: consumer: checkpoint: mode: MANUAL poller: initial-delay: 0 fixed-delay: 1000Nota
Il precedente file di applicazione mostra come configurare un unico poller predefinito da applicare a tutti i binding. Se si vuole configurare il poller per un'associazione specifica, è possibile usare una configurazione come
spring.cloud.stream.bindings.<binding-name>.producer.poller.fixed-delay=3000.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.
Abbiamo bisogno di definire due fornitori e due consumatori:
@Bean public Supplier<Message<String>> supply1() { return () -> { LOGGER.info("Sending message1, sequence1 " + i); return MessageBuilder.withPayload("Hello world1, " + i++).build(); }; } @Bean public Supplier<Message<String>> supply2() { return () -> { LOGGER.info("Sending message2, sequence2 " + j); return MessageBuilder.withPayload("Hello world2, " + j++).build(); }; } @Bean public Consumer<Message<String>> consume1() { return message -> { Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER); LOGGER.info("New message1 received: '{}'", message); checkpointer.success() .doOnSuccess(success -> LOGGER.info("Message1 '{}' successfully checkpointed", message)) .doOnError(error -> LOGGER.error("Exception found", error)) .block(); }; } @Bean public Consumer<Message<String>> consume2() { return message -> { Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER); LOGGER.info("New message2 received: '{}'", message); checkpointer.success() .doOnSuccess(success -> LOGGER.info("Message2 '{}' successfully checkpointed", message)) .doOnError(error -> LOGGER.error("Exception found", error)) .block(); }; }
Allocazione delle risorse
Il binding di Event Hubs supporta il provisioning dell'hub eventi e del gruppo consumer. Gli utenti possono usare le proprietà seguenti per abilitare il provisioning.
spring:
cloud:
azure:
credential:
tenant-id: <tenant>
profile:
subscription-id: ${AZURE_SUBSCRIPTION_ID}
eventhubs:
resource:
resource-group: ${AZURE_EVENTHUBS_RESOURCE_GROUP}
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.
Campioni
Per altre informazioni, vedere il azure-spring-boot-samples repository in GitHub.
Spring Cloud Stream Binder per il bus di servizio di Azure
Concetti chiave
Spring Cloud Stream Binder per bus di servizio di Azure fornisce l'implementazione del binding per il framework Spring Cloud Stream. Questa implementazione si basa sugli adattatori di canale di Spring Integration bus di servizio.
Messaggio pianificato
Questo binder consente l'invio di messaggi a un topic per l'elaborazione differita. Gli utenti possono inviare messaggi programmati con l'intestazione x-delay, che esprime in millisecondi un tempo di ritardo per il messaggio. Il messaggio verrà recapitato ai rispettivi argomenti dopo x-delay millisecondi.
Gruppo di consumatori
Il topic di bus di servizio offre un supporto ai gruppi di consumer simile a quello di Apache Kafka, ma con una logica leggermente diversa.
Questo binder fa affidamento su Subscription di un topic per fungere da gruppo di consumatori.
Configurazione delle dipendenze
<dependency>
<groupId>com.azure.spring</groupId>
<artifactId>spring-cloud-azure-stream-binder-servicebus</artifactId>
</dependency>
In alternativa, è anche possibile usare Spring Cloud Azure Stream bus di servizio Starter, come illustrato nell'esempio seguente per Maven:
<dependency>
<groupId>com.azure.spring</groupId>
<artifactId>spring-cloud-azure-starter-stream-servicebus</artifactId>
</dependency>
Configurazione
Il binder 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
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-stream-binder-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. |
Nota
Le opzioni di configurazione comuni dell’SDK Common Azure Service possono essere configurate anche per il binder Spring Cloud Azure Stream bus di servizio. 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.servicebus..
Il binder supporta anche Spring Could Azure Resource Manager per impostazione predefinita. Per informazioni su come recuperare la stringa di connessione con principal di sicurezza a cui non sono stati concessi i ruoli correlati a Data, vedere la sezione Utilizzo di base di Spring Cloud Azure Resource Manager.
Proprietà di configurazione dell'associazione del bus di servizio di Azure
Le opzioni seguenti sono suddivise in quattro sezioni: Proprietà del consumatore, Configurazioni avanzate del consumatore, Proprietà del produttore e Configurazioni avanzate del produttore.
Proprietà del consumatore
Queste proprietà vengono esposte tramite ServiceBusConsumerProperties.
Nota
Per evitare ripetizioni, a partire dalle versioni 4.17.0 e 5.11.0, Spring Cloud Azure Stream Binder bus di servizio supporta l'impostazione di valori per tutti i canali, nel formato spring.cloud.stream.servicebus.default.consumer.<property>=<value>.
Proprietà di spring-cloud-azure-stream-binder-servicebus configurabili dall'utente:
| Proprietà | Digitare | Impostazione predefinita | Descrizione |
|---|---|---|---|
spring.cloud.stream.servicebus.bindings.binding-name.consumer.requeue-rejected |
booleano | falso | Se i messaggi non recapitati vengono instradati alla DLQ. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.max-concurrent-calls |
Numero intero | 1 | Numero massimo di messaggi simultanei che il client processore del bus di servizio deve elaborare. Quando la sessione è abilitata, si applica a ogni sessione. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.max-concurrent-sessions |
Numero intero | nullo | Numero massimo di sessioni simultanee da elaborare in un determinato momento. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.session-enabled |
Booleano | nullo | Indica se la sessione è abilitata. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.session-idle-timeout |
Durata | nullo | Imposta la quantità massima di tempo (durata) per attendere la ricezione di un messaggio per la sessione attualmente attiva. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.prefetch-count |
Numero intero | 0 | Numero di elementi preletti del client del processore di bus di servizio. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.sub-queue |
Accodamento secondario | nessuno | Tipo della coda secondaria a cui connettersi. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.max-auto-lock-renew-duration |
Durata | 5 milioni | Quantità di tempo per continuare il rinnovo automatico del blocco. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.receive-mode |
ServiceBusReceiveMode | peek_lock | Modalità di ricezione del client processore di bus di servizio. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.auto-complete |
Booleano | vero | Indica se risolvere automaticamente i messaggi. Se impostato su false, verrà aggiunta un'intestazione di messaggio di Checkpointer per consentire agli sviluppatori di risolvere manualmente i messaggi. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.max-size-in-megabytes |
Lungo | 1024 | Dimensioni massime della coda o dell'argomento in megabyte, ovvero la dimensione della memoria allocata per la coda o l'argomento. |
spring.cloud.stream.servicebus.bindings.binding-name.consumer.default-message-time-to-live |
Durata | P10675199DT2H48M5.4775807S. (10675199 giorni, 2 ore, 48 minuti, 5 secondi e 477 millisecondi) | Durata dopo la quale il messaggio scade, a partire da quando il messaggio viene inviato al bus di servizio. |
Importante
Quando si usa Azure Resource Manager (ARM), è necessario configurare la proprietà spring.cloud.stream.servicebus.bindings.<binding-name>.consume.entity-type. Per altre informazioni, vedere l'esempio servicebus-queue-binder-arm in GitHub.
Configurazione avanzata utente
La configurazione della connessione precedente e del client comune di Azure SDK supporta la personalizzazione per ogni consumer del binder, che è possibile configurare con il prefisso spring.cloud.stream.servicebus.bindings.<binding-name>.consumer..
Proprietà del produttore
Queste proprietà vengono esposte tramite ServiceBusProducerProperties.
Nota
Per evitare ripetizioni, a partire dalle versioni 4.17.0 e 5.11.0, Spring Cloud Azure Stream Binder bus di servizio supporta l'impostazione di valori per tutti i canali, nel formato spring.cloud.stream.servicebus.default.producer.<property>=<value>.
Proprietà di spring-cloud-azure-stream-binder-servicebus configurabili dal produttore:
| Proprietà | Digitare | Impostazione predefinita | Descrizione |
|---|---|---|---|
spring.cloud.stream.servicebus.bindings.binding-name.producer.sync |
booleano | falso | Interruttore per la sincronizzazione del produttore. |
spring.cloud.stream.servicebus.bindings.binding-name.producer.send-timeout |
lungo | 10.000 | Valore di timeout per l'invio del producer. |
spring.cloud.stream.servicebus.bindings.binding-name.producer.entity-type |
ServiceBusEntityType | nullo | Tipo di entità di bus di servizio del produttore, necessario per il produttore di binding. |
spring.cloud.stream.servicebus.bindings.binding-name.producer.max-size-in-megabytes |
Lungo | 1024 | Dimensioni massime della coda o dell'argomento in megabyte, ovvero la dimensione della memoria allocata per la coda o l'argomento. |
spring.cloud.stream.servicebus.bindings.binding-name.producer.default-message-time-to-live |
Durata | P10675199DT2H48M5.4775807S. (10675199 giorni, 2 ore, 48 minuti, 5 secondi e 477 millisecondi) | Durata dopo la quale il messaggio scade, a partire da quando il messaggio viene inviato al bus di servizio. |
Importante
Quando si usa il producer di binding, è necessario configurare la proprietà spring.cloud.stream.servicebus.bindings.<binding-name>.producer.entity-type.
Configurazione avanzata del produttore
La configurazione precedente di connessione e del client comune di Azure SDK supporta la personalizzazione per ogni producer del binder, che è possibile configurare con il prefisso spring.cloud.stream.servicebus.bindings.<binding-name>.producer..
Utilizzo di base
Invio e ricezione di messaggi da/al bus di servizio
Compilare le opzioni di configurazione con le informazioni sulle credenziali.
Per le credenziali come stringa di connessione, configurare le proprietà seguenti nel file application.yml:
spring: cloud: azure: servicebus: connection-string: ${SERVICEBUS_NAMESPACE_CONNECTION_STRING} function: definition: consume;supply stream: bindings: consume-in-0: destination: ${SERVICEBUS_ENTITY_NAME} # If you use Service Bus Topic, add the following configuration # group: ${SUBSCRIPTION_NAME} supply-out-0: destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE} servicebus: bindings: consume-in-0: consumer: auto-complete: false supply-out-0: producer: entity-type: queue # set as "topic" if you use Service Bus TopicNota
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 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: ${SERVICEBUS_NAMESPACE} function: definition: consume;supply stream: bindings: consume-in-0: destination: ${SERVICEBUS_ENTITY_NAME} # If you use Service Bus Topic, add the following configuration # group: ${SUBSCRIPTION_NAME} supply-out-0: destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE} servicebus: bindings: consume-in-0: consumer: auto-complete: false supply-out-0: producer: entity-type: queue # set as "topic" if you use Service Bus Topic
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.
Per le credenziali come identità gestite, configurare le proprietà seguenti nel file application.yml:
spring: cloud: azure: credential: managed-identity-enabled: true client-id: ${MANAGED_IDENTITY_CLIENT_ID} # Only needed when using a user-assigned managed identity servicebus: namespace: ${SERVICEBUS_NAMESPACE} function: definition: consume;supply stream: bindings: consume-in-0: destination: ${SERVICEBUS_ENTITY_NAME} # If you use Service Bus Topic, add the following configuration # group: ${SUBSCRIPTION_NAME} supply-out-0: destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE} servicebus: bindings: consume-in-0: consumer: auto-complete: false supply-out-0: producer: entity-type: queue # set as "topic" if you use Service Bus Topic
Definire i fornitori e i consumatori.
@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(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload())) .doOnError(error -> LOGGER.error("Exception found", error)) .block(); }; } @Bean public Supplier<Message<String>> supply() { return () -> { LOGGER.info("Sending message, sequence " + i); return MessageBuilder.withPayload("Hello world, " + i++).build(); }; }
Supporto per la chiave di partizione
Il binder 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.
Spring Cloud Stream fornisce una proprietà dell'espressione SpEL della chiave di partizione spring.cloud.stream.bindings.<binding-name>.producer.partition-key-expression. Ad esempio, impostando questa proprietà come "'partitionKey-' + headers[<message-header-key>]" e aggiungendo un'intestazione denominata message-header-key. Spring Cloud Stream usa il valore di questa intestazione quando valuta l'espressione per determinare una chiave di partizione. Il codice seguente mostra un produttore di esempio:
@Bean
public Supplier<Message<String>> generate() {
return () -> {
String value = "random payload";
return MessageBuilder.withPayload(value)
.setHeader("<message-header-key>", value.length() % 4)
.build();
};
}
Supporto delle sessioni
Il binder supporta le sessioni di messaggi di bus di servizio. L'ID sessione di un messaggio può essere impostato tramite l'intestazione del messaggio.
@Bean
public Supplier<Message<String>> generate() {
return () -> {
String value = "random payload";
return MessageBuilder.withPayload(value)
.setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session ID")
.build();
};
}
Nota
In base a partizionamento del bus di servizio, l'ID sessione ha una priorità più alta rispetto alla chiave di partizione. Pertanto, quando vengono impostate entrambe le intestazioni ServiceBusMessageHeaders#SESSION_ID e ServiceBusMessageHeaders#PARTITION_KEY, il valore dell'ID sessione viene infine usato per sovrascrivere il valore della chiave di partizione.
Gestire i messaggi di errore
- Spring Cloud Azure 6.x/7.x
- Spring Cloud Azure 5.x
- Spring Cloud Azure 4.x
Gestire i messaggi di errore di associazione in uscita
Per impostazione predefinita, Spring Integration crea un canale di errore globale denominato
errorChannel. Configurare l'endpoint del messaggio seguente per gestire il messaggio di errore di associazione in uscita.@ServiceActivator(inputChannel = IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) public void handleError(ErrorMessage message) { LOGGER.error("Handling outbound binding error: " + message); }Gestire i messaggi di errore di associazione in ingresso
Spring Cloud Stream bus di servizio Binder supporta due soluzioni per gestire gli errori per le associazioni di messaggi in ingresso: il gestore degli errori del binder e i gestori.
Gestore degli errori Binder:
Il gestore di errori predefinito del binder gestisce il binding in ingresso. Si usa questo gestore per inviare i messaggi non inviati correttamente alla coda dei messaggi non recapitabili quando
spring.cloud.stream.servicebus.bindings.<binding-name>.consumer.requeue-rejectedè abilitato. In caso contrario, i messaggi non riusciti vengono abbandonati. Il gestore degli errori del binder è mutuamente esclusivo rispetto agli altri gestori degli errori forniti.gestore degli errori:
Spring Cloud Stream espone un meccanismo per fornire un gestore degli errori personalizzato aggiungendo un
Consumerche accetta istanze diErrorMessage. Per altre informazioni, vedere Gestire i messaggi di errore nella documentazione di Spring Cloud Stream.Gestore di errori predefinito per il binding
Configurare un singolo bean
Consumerper gestire tutti i messaggi di errore del binding in ingresso. La seguente funzione predefinita si sottoscrive a ciascun canale di errore dei binding in ingresso:@Bean public Consumer<ErrorMessage> myDefaultHandler() { return message -> { // consume the error message }; }È anche necessario impostare la proprietà
spring.cloud.stream.default.error-handler-definitionsul nome della funzione.Gestore degli errori specifico del binding
Configurare un
Consumerbean per gestire gli specifici messaggi di errore del binding in ingresso. La funzione seguente si registra al canale di errore del binding di ingresso specifico con una priorità più alta rispetto al gestore errori predefinito del binding.@Bean public Consumer<ErrorMessage> myDefaultHandler() { return message -> { // consume the error message }; }È anche necessario impostare la proprietà
spring.cloud.stream.bindings.<input-binding-name>.error-handler-definitionsul nome della funzione.
Intestazioni dei messaggi di bus di servizio
Per le intestazioni di base dei messaggi supportate, vedere la sezione intestazioni dei messaggi di bus di servizio di Supporto di Spring Cloud Azure per Spring Integration.
Nota
Quando si imposta la chiave di partizione, la priorità dell'intestazione del messaggio è superiore alla proprietà Spring Cloud Stream. Pertanto, spring.cloud.stream.bindings.<binding-name>.producer.partition-key-expression diventa effettivo solo quando non è configurata nessuna delle intestazioni ServiceBusMessageHeaders#SESSION_ID e ServiceBusMessageHeaders#PARTITION_KEY.
Supporto di più binder
È supportata anche la connessione a più namespace di bus di servizio tramite l'uso di binder multipli. Questo esempio utilizza la stringa di connessione come esempio. Sono supportate anche le credenziali delle entità servizio e delle identità gestite. Gli utenti possono impostare proprietà correlate nelle impostazioni dell'ambiente di ogni binder.
Per usare più binder di ServiceBus, configurare le proprietà seguenti nel file application.yml:
spring: cloud: function: definition: consume1;supply1;consume2;supply2 stream: bindings: consume1-in-0: destination: ${SERVICEBUS_TOPIC_NAME} group: ${SUBSCRIPTION_NAME} supply1-out-0: destination: ${SERVICEBUS_TOPIC_NAME_SAME_AS_ABOVE} consume2-in-0: binder: servicebus-2 destination: ${SERVICEBUS_QUEUE_NAME} supply2-out-0: binder: servicebus-2 destination: ${SERVICEBUS_QUEUE_NAME_SAME_AS_ABOVE} binders: servicebus-1: type: servicebus default-candidate: true environment: spring: cloud: azure: servicebus: connection-string: ${SERVICEBUS_NAMESPACE_01_CONNECTION_STRING} servicebus-2: type: servicebus default-candidate: false environment: spring: cloud: azure: servicebus: connection-string: ${SERVICEBUS_NAMESPACE_02_CONNECTION_STRING} servicebus: bindings: consume1-in-0: consumer: auto-complete: false supply1-out-0: producer: entity-type: topic consume2-in-0: consumer: auto-complete: false supply2-out-0: producer: entity-type: queue poller: initial-delay: 0 fixed-delay: 1000Nota
Il precedente file di applicazione mostra come configurare un unico poller predefinito da applicare a tutti i binding. Se si vuole configurare il poller per un'associazione specifica, è possibile usare una configurazione come
spring.cloud.stream.bindings.<binding-name>.producer.poller.fixed-delay=3000.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.
abbiamo bisogno di definire due fornitori e due consumatori
@Bean public Supplier<Message<String>> supply1() { return () -> { LOGGER.info("Sending message1, sequence1 " + i); return MessageBuilder.withPayload("Hello world1, " + i++).build(); }; } @Bean public Supplier<Message<String>> supply2() { return () -> { LOGGER.info("Sending message2, sequence2 " + j); return MessageBuilder.withPayload("Hello world2, " + j++).build(); }; } @Bean public Consumer<Message<String>> consume1() { return message -> { Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER); LOGGER.info("New message1 received: '{}'", message); checkpointer.success() .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload())) .doOnError(e -> LOGGER.error("Error found", e)) .block(); }; } @Bean public Consumer<Message<String>> consume2() { return message -> { Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER); LOGGER.info("New message2 received: '{}'", message); checkpointer.success() .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload())) .doOnError(e -> LOGGER.error("Error found", e)) .block(); }; }
Allocazione delle risorse
Il binder del bus di servizio supporta il provisioning di code, topic e sottoscrizioni. Gli utenti possono usare le proprietà seguenti per attivare il provisioning.
spring:
cloud:
azure:
credential:
tenant-id: <tenant>
profile:
subscription-id: ${AZURE_SUBSCRIPTION_ID}
servicebus:
resource:
resource-group: ${AZURE_SERVICEBUS_RESOURCE_GROUP}
stream:
servicebus:
bindings:
<binding-name>:
consumer:
entity-type: ${SERVICEBUS_CONSUMER_ENTITY_TYPE}
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.
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.