Podpora Spring Cloud Azure pro integraci Spring

Rozšíření Spring Integration pro Azure poskytuje adaptéry integrace Spring pro různé služby poskytované sadou Azure SDK pro Javu. Poskytujeme podporu pro Spring Integration pro tyto služby Azure: Event Hubs, Service Bus, Storage Queue. Následuje seznam podporovaných adaptérů:

• spring-cloud-azure-starter-integration-eventhubs – další informace najdete v tématu Integrace Spring se službou Azure Event Hubs
• spring-cloud-azure-starter-integration-servicebus – další informace najdete v tématu Integrace Spring se službou Azure Service Bus
• spring-cloud-azure-starter-integration-storage-queue - další informace najdete v článku Spring Integration with Azure Storage Queue

Integrace Spring se službou Azure Event Hubs

Klíčové koncepty

Azure Event Hubs je platforma pro streamování velkých objemů dat a služba pro příjem událostí. Může přijímat a zpracovávat miliony událostí za sekundu. Data odesílaná do centra událostí je možné transformovat a ukládat pomocí libovolného poskytovatele analýz v reálném čase nebo adaptérů batchingu nebo úložiště.

Spring Integration umožňuje zjednodušené zasílání zpráv v aplikacích založených na Springu a podporuje integraci s externími systémy prostřednictvím deklarativních adaptérů. Tyto adaptéry poskytují vyšší úroveň abstrakce nad podporou vzdálené komunikace, zasílání zpráv a plánování ve Springu. Projekt rozšíření Spring Integration pro Event Hubs poskytuje adaptéry a brány vstupních a výstupních kanálů pro službu Azure Event Hubs.

Poznámka

Rozhraní API pro podporu RxJava jsou ve verzi 4.0.0 odstraněna. Podrobnosti najdete v Javadocu.

Skupina spotřebitelů

Event Hubs poskytuje podobnou podporu skupiny příjemců jako Apache Kafka, ale s mírnou odlišnou logikou. Zatímco Kafka ukládá všechny potvrzené offsety v brokeru, offsety zpráv služby Event Hubs, které se zpracovávají, musíte ukládat ručně. Sada Event Hubs SDK poskytuje funkci pro ukládání takových posunů ve službě Azure Storage.

Podpora rozdělení na oddíly

Služba Event Hubs poskytuje podobný koncept fyzického oddílu jako Kafka. Na rozdíl od automatického vyrovnávání mezi uživateli a oddíly Kafka ale služba Event Hubs poskytuje druh preemptivního režimu. Účet úložiště funguje jako zapůjčení k určení, který oddíl vlastní uživatel. Když se spustí nový příjemce, pokusí se ukrást některé oddíly od většiny náročných příjemců, aby se dosáhlo vyrovnávání zatížení.

Pokud chcete určit strategii vyrovnávání zatížení, můžou vývojáři pro konfiguraci použít EventHubsContainerProperties. Příklad konfigurace prvku EventHubsContainerProperties naleznete v následující části.

Podpora příjemců služby Batch

EventHubsInboundChannelAdapter podporuje režim dávkového využívání. Pokud ho chcete povolit, mohou uživatelé při vytváření instance ListenerMode.BATCH určit režim naslouchacího procesu jako EventHubsInboundChannelAdapter. Je-li tato možnost povolena, bude přijata zpráva, jejíž datová část je seznamem dávkově zpracovaných událostí, a předána do následného kanálu. Každé záhlaví zprávy je také převedeno jako seznam, jehož obsah je přidružená hodnota záhlaví parsovaná z každé události. U společných hlaviček ID oddílu, checkpointeru a vlastností posledního zařazení do fronty se tyto hodnoty uvádějí jako jediná hodnota pro celou dávku, pokud ji sdílejí všechny události. Další informace najdete v části Hlavičky zpráv služby Event Hubs.

Poznámka

Hlavička kontrolního bodu existuje pouze v případech, kdy se používá režim kontrolního bodu MANUAL.

Vytváření kontrolních bodů dávkového příjemce podporuje dva režimy: BATCH a MANUAL. BATCH režim je režim automatického vytváření kontrolních bodů, ve kterém se po přijetí vytvoří kontrolní bod pro celou dávku událostí najednou. MANUAL režim slouží k vytvoření kontrolního bodu událostí uživateli. Při použití bude Checkpointer předán v hlavičce zprávy a uživatelé jej mohou použít k vytváření kontrolních bodů.

Zásady dávkového využívání lze určit vlastnostmi max-size a max-wait-time, kde max-size je nezbytná vlastnost, zatímco max-wait-time je nepovinná. Pokud chcete určit strategii dávkového využívání, můžou vývojáři pro konfiguraci použít EventHubsContainerProperties. Příklad konfigurace EventHubsContainerProperties najdete v následující části.

Nastavení závislostí

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

Konfigurace

Tato úvodní sada poskytuje následující tři části možností konfigurace:

Vlastnosti konfigurace připojení

Tato část obsahuje možnosti konfigurace používané pro připojení ke službě Azure Event Hubs.

Poznámka

Pokud se rozhodnete použít objekt zabezpečení k ověření a autorizaci pomocí Microsoft Entra ID pro přístup k prostředku Azure, přečtěte si téma Autorizace přístupu s ID Microsoft Entra, abyste měli jistotu, že objekt zabezpečení má dostatečná oprávnění pro přístup k prostředku Azure.

Konfigurovatelné vlastnosti spring-cloud-azure-starter-integration-eventhubspřipojení:

Vlastnost	Typ	Popis
spring.cloud.azure.eventhubs.enabled	booleovský	Určuje, jestli je služba Azure Event Hubs povolená.
spring.cloud.azure.eventhubs.connection-string	Řetězec	Hodnota připojovacího řetězce pro obor názvů služby Event Hubs
spring.cloud.azure.eventhubs.namespace	Řetězec	Hodnota oboru názvů služby Event Hubs, tedy předpona plně kvalifikovaného názvu domény. FQDN by se měl skládat z NamespaceName.DomainName
spring.cloud.azure.eventhubs.domain-name	Řetězec	Název domény pro hodnotu oboru názvů služby Azure Event Hubs
spring.cloud.azure.eventhubs.custom-endpoint-address	Řetězec	Adresa vlastního koncového bodu
spring.cloud.azure.eventhubs.shared-connection	logický	Určuje, jestli základní EventProcessorClient a EventHubProducerAsyncClient používají stejné připojení. Ve výchozím nastavení se pro každého vytvořeného klienta Event Hub vytvoří a použije nové připojení.

Vlastnosti konfigurace kontrolního bodu

Tato část obsahuje možnosti konfigurace pro službu Storage Blobs, která se používá k zachování vlastnictví oddílu a informací kontrolních bodů.

Poznámka

Pokud není vlastnost spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists ručně povolená ve verzi 4.0.0, automaticky se nevytvořil žádný kontejner úložiště.

Konfigurovatelné vlastnosti kontrolních spring-cloud-azure-starter-integration-eventhubsbodů:

Vlastnost	Typ	Popis
spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists	logický	Zda chcete povolit vytváření kontejnerů, pokud neexistuje.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name	Řetězec	Název účtu úložiště
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key	Řetězec	Přístupový klíč účtu úložiště.
spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name	Řetězec	Název kontejneru úložiště

Běžné možnosti konfigurace sady SDK služby Azure lze použít také pro úložiště kontrolních bodů pro objekty blob ve službě Azure Storage. Podporované možnosti konfigurace jsou zavedeny v Spring Cloud Azure konfiguracea lze je nakonfigurovat pomocí sjednocené předpony spring.cloud.azure. nebo předpony spring.cloud.azure.eventhubs.processor.checkpoint-store..

Vlastnosti konfigurace procesoru centra událostí

EventHubsInboundChannelAdapter používá EventProcessorClient ke zpracování zpráv z centra událostí, ke konfiguraci celkových vlastností EventProcessorClientmohou vývojáři použít EventHubsContainerProperties pro konfiguraci. Přečtěte si následující části o tom, jak pracovat s EventHubsInboundChannelAdapter.

Základní využití

Odesílání zpráv do služby Azure Event Hubs

1. Vyplňte možnosti konfigurace přihlašovacích údajů.

   • Pro přihlašovací údaje jako připojovací řetězec nakonfigurujte v souboru application.yml následující vlastnosti:

     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}
     

     Poznámka

     Microsoft doporučuje používat nejbezpečnější dostupný tok ověřování. Tok ověřování popsaný v tomto postupu, například pro databáze, mezipaměti, zasílání zpráv nebo služby AI, vyžaduje velmi vysoký stupeň důvěryhodnosti v aplikaci a nese rizika, která nejsou přítomna v jiných tocích. Tento postup používejte pouze tehdy, pokud bezpečnější možnosti, jako jsou spravované identity pro připojení bez hesla nebo klíče, nejsou vhodné. V případě místních operací počítačů upřednostňujete identity uživatelů pro připojení bez hesla nebo bez klíčů.

   • U přihlašovacích údajů jako spravovaných identit nakonfigurujte v souboru application.yml následující vlastnosti:

     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}
     

   • Pro přihlašovací údaje pro instanční objekt služby nakonfigurujte v souboru application.yml následující vlastnosti:

     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}
     

Poznámka

Hodnoty povolené pro tenant-id jsou: common, organizations, consumersnebo ID tenanta. Další informace o těchto hodnotách najdete v části Byl použit nesprávný koncový bod (osobní účty a účty organizací) v článku Chyba AADSTS50020 – Uživatelský účet od poskytovatele identity v tenantovi neexistuje. Informace o převodu aplikace s jedním tenantem najdete v tématu Převod jednoklientských aplikací na víceklienta vMicrosoft Entra ID .

1. Vytvořte DefaultMessageHandler s využitím beanu EventHubsTemplate k odesílání zpráv do 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;
       }
   }
   

2. Vytvořte propojení brány zpráv s výše uvedenou obslužnou rutinou zpráv prostřednictvím kanálu zpráv.

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

3. Odesílání zpráv pomocí brány

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

Příjem zpráv ze služby Azure Event Hubs

1. Vyplňte možnosti konfigurace přihlašovacích údajů.

2. Vytvořte bean kanálu zpráv jako vstupní kanál.

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

3. Vytvořte EventHubsInboundChannelAdapter pomocí beanu EventHubsMessageListenerContainer pro příjem zpráv ze služby 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);
       }
   }
   

4. Vytvořte vazbu přijímače zpráv pomocí EventHubsInboundChannelAdapter prostřednictvím dříve vytvořeného kanálu zpráv.

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

Konfigurace EventHubsMessageConverter pro přizpůsobení objectMapper

EventHubsMessageConverter se vytvoří jako konfigurovatelná bean, která uživatelům umožní přizpůsobit ObjectMapper.

Podpora příjemců služby Batch

Pokud chcete využívat zprávy ze služby Event Hubs v dávkách, je podobné jako u výše uvedené ukázky, kromě toho by uživatelé měli nastavit možnosti konfigurace související s dávkovou spotřebou pro EventHubsInboundChannelAdapter.

Při vytváření EventHubsInboundChannelAdapterby měl být režim naslouchacího procesu nastaven jako BATCH. Při vytváření beanu EventHubsMessageListenerContainer nastavte režim checkpointu buď na MANUAL, nebo na BATCH, a volby dávky můžete podle potřeby nakonfigurovat.

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

Hlavičky zpráv služby Event Hubs

Následující tabulka ukazuje, jak jsou vlastnosti zpráv služby Event Hubs mapovány na hlavičky zpráv Spring. U služby Azure Event Hubs se zpráva volá jako event.

Mapování mezi zprávou / vlastnostmi události ve službě Event Hubs a záhlavími zpráv Springu v režimu naslouchání záznamů:

Vlastnosti událostí služby Event Hubs	Konstanty záhlaví springových zpráv	Typ	Popis
Čas zařazení do fronty	EventHubsHeaders#ENQUEUED_TIME	Okamžitý	Okamžik v čase UTC, kdy byla událost zařazena do fronty v oddílu služby Event Hubs.
Posun	EventHubsHeaders#OFFSET	Dlouhý	Offset události v okamžiku jejího přijetí z přidruženého oddílu Event Hubu.
Klíč partition	AzureHeaders#PARTITION_KEY	Řetězec	Hašovací klíč partition, pokud byl nastaven při původním publikování této události.
ID oddílu	AzureHeaders#RAW_PARTITION_ID	Řetězec	ID oddílu služby Event Hub.
Pořadové číslo	EventHubsHeaders#SEQUENCE_NUMBER	Dlouhý	Pořadové číslo přidělené události při jejím zařazení do fronty v přidruženém oddílu Event Hubu.
Vlastnosti poslední události ve frontě	EventHubsHeaders#LAST_ENQUEUED_EVENT_PROPERTIES	Vlastnosti události LastEnqueuedEvent.	Vlastnosti poslední události v tomto oddílu.
NA	AzureHeaders#CHECKPOINTER	Kontrolní bod	Záhlaví pro kontrolu konkrétní zprávy.

Uživatelé mohou analyzovat záhlaví zprávy pro související informace o každé události. Pokud chcete nastavit záhlaví zprávy pro událost, budou všechna přizpůsobená záhlaví vložena jako vlastnost aplikace události, kde je hlavička nastavena jako klíč vlastnosti. Při přijetí událostí ze služby Event Hubs se všechny vlastnosti aplikace převedou do záhlaví zprávy.

Poznámka

Není podporováno ruční nastavení záhlaví zpráv, jako jsou klíč oddílu, čas zařazení do fronty, offset a pořadové číslo.

Pokud je povolený režim dávkového příjemce, zobrazí se následující konkrétní hlavičky dávkových zpráv, které obsahují seznam hodnot z každé události služby Event Hubs.

Mapování mezi vlastnostmi zpráv a událostí služby Event Hubs a hlavičkami zpráv Springu v režimu dávkového naslouchání:

Vlastnosti událostí služby Event Hubs	Konstanty záhlaví zpráv Spring Batch	Typ	Popis
Čas zařazení do fronty	EventHubsHeaders#ENQUEUED_TIME	Seznam Instant	Seznam časových okamžiků v UTC, kdy byla jednotlivá událost zařazena do fronty v oddílu služby Event Hubs.
Posun	EventHubsHeaders#OFFSET	Seznam dlouhých	Seznam offsetů pro každou událost v okamžiku jejího přijetí z přidruženého oddílu služby Event Hub.
Klíč partition	AzureHeaders#PARTITION_KEY	Seznam řetězců	Uveďte hashovací klíč oddílu, pokud byl nastaven při původním publikování jednotlivých událostí.
Pořadové číslo	EventHubsHeaders#SEQUENCE_NUMBER	Seznam dlouhých	Seznam pořadových čísel přiřazených jednotlivým událostem při jejich zařazení do fronty v přidruženém oddílu Event Hubu.
Systémové vlastnosti	Hlavičky EventHubs#BATCH_CONVERTED_SYSTEM_PROPERTIES	Seznam map	Seznam systémových vlastností každé události
Vlastnosti aplikace	Hlavičky EventHubs#BATCH_CONVERTED_APPLICATION_PROPERTIES	Seznam map	Seznam vlastností aplikace každé události, kde jsou umístěny všechny přizpůsobené hlavičky zpráv nebo vlastnosti události.

Poznámka

Při publikování zpráv budou všechny výše uvedené dávkové hlavičky odebrány ze zpráv, pokud existují.

Vzorky

Další informace najdete v úložišti na GitHubu azure-spring-boot-samples .

Integrace Spring se službou Azure Service Bus

Klíčové koncepty

Spring Integration umožňuje zjednodušené zasílání zpráv v aplikacích založených na Springu a podporuje integraci s externími systémy prostřednictvím deklarativních adaptérů.

Projekt rozšíření Spring Integration for Azure Service Bus poskytuje adaptéry příchozích a odchozích kanálů pro Službu Azure Service Bus.

Poznámka

Rozhraní API pro podporu CompletableFuture jsou od verze 2.10.0 označena jako zastaralá a od verze 4.0.0 jsou nahrazena nástrojem Reactor Core. Podrobnosti najdete v Javadocu.

Nastavení závislostí

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

Konfigurace

Tato úvodní sada poskytuje následující dvě části možností konfigurace:

Vlastnosti konfigurace připojení

Tato část obsahuje možnosti konfigurace používané pro připojení ke službě Azure Service Bus.

Poznámka

Pokud se rozhodnete použít objekt zabezpečení k ověření a autorizaci pomocí Microsoft Entra ID pro přístup k prostředku Azure, přečtěte si téma Autorizace přístupu s ID Microsoft Entra, abyste měli jistotu, že objekt zabezpečení má dostatečná oprávnění pro přístup k prostředku Azure.

Konfigurovatelné vlastnosti spring-cloud-azure-starter-integration-servicebuspřipojení:

Vlastnost	Typ	Popis
spring.cloud.azure.servicebus.enabled	booleovský	Určuje, jestli je povolená služba Azure Service Bus.
spring.cloud.azure.servicebus.connection-string	Řetězec	Hodnota připojovacího řetězce pro obor názvů služby Service Bus
spring.cloud.azure.servicebus.custom-endpoint-address	Řetězec	Vlastní adresa koncového bodu, která se má použít při připojování ke službě Service Bus.
spring.cloud.azure.servicebus.namespace	Řetězec	Hodnota oboru názvů služby Service Bus, což je předpona plně kvalifikovaného názvu domény. FQDN by se měl skládat z NamespaceName.DomainName
spring.cloud.azure.servicebus.domain-name	Řetězec	Název domény hodnoty oboru názvů služby Azure Service Bus

Vlastnosti konfigurace procesoru služby Service Bus

ServiceBusInboundChannelAdapter používá ServiceBusProcessorClient ke zpracování zpráv, ke konfiguraci celkových vlastností ServiceBusProcessorClient, mohou vývojáři pro konfiguraci použít ServiceBusContainerProperties. Přečtěte si následující části o tom, jak pracovat s ServiceBusInboundChannelAdapter.

Základní využití

Odesílání zpráv do služby Azure Service Bus

1. Vyplňte možnosti konfigurace přihlašovacích údajů.

   • Pro přihlašovací údaje jako připojovací řetězec nakonfigurujte v souboru application.yml následující vlastnosti:

     spring:
       cloud:
         azure:
           servicebus:
             connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}
     

     Poznámka

     Microsoft doporučuje používat nejbezpečnější dostupný tok ověřování. Tok ověřování popsaný v tomto postupu, například pro databáze, mezipaměti, zasílání zpráv nebo služby AI, vyžaduje velmi vysoký stupeň důvěryhodnosti v aplikaci a nese rizika, která nejsou přítomna v jiných tocích. Tento postup používejte pouze tehdy, pokud bezpečnější možnosti, jako jsou spravované identity pro připojení bez hesla nebo klíče, nejsou vhodné. V případě místních operací počítačů upřednostňujete identity uživatelů pro připojení bez hesla nebo bez klíčů.

   • U přihlašovacích údajů jako spravovaných identit nakonfigurujte v souboru application.yml následující vlastnosti:

     spring:
       cloud:
         azure:
           credential:
             managed-identity-enabled: true
             client-id: ${AZURE_CLIENT_ID}
           profile:
             tenant-id: <tenant>
           servicebus:
             namespace: ${AZURE_SERVICE_BUS_NAMESPACE}
     

Poznámka

Hodnoty povolené pro tenant-id jsou: common, organizations, consumersnebo ID tenanta. Další informace o těchto hodnotách najdete v části Byl použit nesprávný koncový bod (osobní účty a účty organizací) v článku Chyba AADSTS50020 – Uživatelský účet od poskytovatele identity v tenantovi neexistuje. Informace o převodu aplikace s jedním tenantem najdete v tématu Převod jednoklientských aplikací na víceklienta vMicrosoft Entra ID .

• Pro přihlašovací údaje pro instanční objekt služby nakonfigurujte v souboru application.yml následující vlastnosti:

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

Poznámka

Hodnoty povolené pro tenant-id jsou: common, organizations, consumersnebo ID tenanta. Další informace o těchto hodnotách najdete v části Byl použit nesprávný koncový bod (osobní účty a účty organizací) v článku Chyba AADSTS50020 – Uživatelský účet od poskytovatele identity v tenantovi neexistuje. Informace o převodu aplikace s jedním tenantem najdete v tématu Převod jednoklientských aplikací na víceklienta vMicrosoft Entra ID .

1. Vytvořte DefaultMessageHandler pomocí ServiceBusTemplate bean pro odesílání zpráv do služby Service Bus a nastavte typ entity pro ServiceBusTemplate. Tato ukázka používá jako příklad frontu služby Service Bus.

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

2. Vytvořte propojení brány zpráv s výše uvedenou obslužnou rutinou zpráv prostřednictvím kanálu zpráv.

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

3. Odesílání zpráv pomocí brány

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

Příjem zpráv ze služby Azure Service Bus

1. Vyplňte možnosti konfigurace přihlašovacích údajů.

2. Vytvořte bean kanálu zpráv jako vstupní kanál.

   @Configuration
   class Demo {
       private static final String INPUT_CHANNEL = "input";
   
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   }
   

3. Vytvořte ServiceBusInboundChannelAdapter pomocí beanu ServiceBusMessageListenerContainer pro příjem zpráv ze služby Service Bus. Tato ukázka používá frontu služby Service Bus jako příklad.

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

4. Vytvořte vazbu příjemce zprávy s ServiceBusInboundChannelAdapter prostřednictvím kanálu zprávy, který jsme vytvořili dříve.

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

Nakonfigurujte ServiceBusMessageConverter pro přizpůsobení objectMapper

ServiceBusMessageConverter se vytvoří jako konfigurovatelná bean, která uživatelům umožní přizpůsobit ObjectMapper.

Hlavičky zpráv služby Service Bus

U některých hlaviček služby Service Bus, které je možné mapovat na více konstant záhlaví Spring, je uvedena priorita různých hlaviček Spring.

Mapování mezi hlavičkami Service Bus a hlavičkami Spring:

Hlavičky a vlastnosti zpráv služby Service Bus	Konstanty hlaviček zpráv ve Springu	Typ	Konfigurovatelný	Popis
Typ obsahu	MessageHeaders#CONTENT_TYPE	Řetězec	Ano	Popisovač typu obsahu zprávy RFC2045.
Identifikátor korelace	ServiceBusMessageHeaders#CORRELATION_ID	Řetězec	Ano	ID korelace zprávy
ID zprávy	ServiceBusMessageHeaders#MESSAGE_ID	Řetězec	Ano	ID dané zprávy, toto záhlaví má vyšší prioritu než MessageHeaders#ID.
ID zprávy	MessageHeaders#ID	Univerzálně jedinečný identifikátor (UUID)	Ano	ID této zprávy, tato hlavička má nižší prioritu než ServiceBusMessageHeaders#MESSAGE_ID.
Klíč partition	ServiceBusMessageHeaders#PARTITION_KEY	Řetězec	Ano	Klíč oddílu pro odeslání zprávy do entity rozdělené do oddílů.
Odpovědět na	MessageHeaders#REPLY_CHANNEL	Řetězec	Ano	Adresa entity, na kterou se mají odesílat odpovědi.
Odpověď na ID relace	ServiceBusMessageHeaders#REPLY_TO_SESSION_ID	Řetězec	Ano	Hodnota vlastnosti ReplyToGroupId zprávy.
Naplánovaný čas zařazení do fronty UTC	ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME	Posun data a času	Ano	Datum a čas, kdy má být zpráva zařazena do fronty ve službě Service Bus. Tato hlavička má vyšší prioritu než AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE.
Naplánovaný čas zařazení do fronty UTC	AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE	Celé číslo	Ano	Datum a čas, kdy má být zpráva zařazena do fronty ve službě Service Bus; tato hlavička má nižší prioritu než ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME.
ID relace	ServiceBusMessageHeaders#SESSION_ID	Řetězec	Ano	Identifikátor relace pro entitu pracující s relací.
Doba životnosti	ServiceBusMessageHeaders#TIME_TO_LIVE	Trvání	Ano	Doba trvání před vypršením platnosti této zprávy.
K	ServiceBusMessageHeaders#TO	Řetězec	Ano	Adresa příjemce zprávy, vyhrazená pro budoucí použití ve scénářích směrování a kterou broker aktuálně sám ignoruje.
Předmět	ServiceBusMessageHeaders#SUBJECT	Řetězec	Ano	Předmět zprávy.
Popis chyby nedoručených zpráv	ServiceBusMessageHeaders#DEAD_LETTER_ERROR_DESCRIPTION	Řetězec	Ne	Popis zprávy, která byla nedoručována.
Důvod nedoručených dopisů	ServiceBusMessageHeaders#DEAD_LETTER_REASON	Řetězec	Ne	Důvod, proč byla zpráva nedoručována.
Zdroj nedoručených dopisů	ServiceBusMessageHeaders#DEAD_LETTER_SOURCE	Řetězec	Ne	Entita, ve které byla zpráva nedoručována.
Počet doručení	ServiceBusMessageHeaders#DELIVERY_COUNT	dlouhý	Ne	Počet doručení této zprávy klientům.
Pořadové číslo ve frontě	ServiceBusMessageHeaders#ENQUEUED_SEQUENCE_NUMBER	dlouhý	Ne	Pořadové číslo přiřazené ke zprávě službou Service Bus.
Čas zařazení do fronty	ServiceBusMessageHeaders#ENQUEUED_TIME	Posun data a času	Ne	Datum a čas, kdy byla tato zpráva zařazena do fronty ve službě Service Bus.
Platnost končí	ServiceBusMessageHeaders#EXPIRES_AT	Posun data a času	Ne	Datum a čas, kdy platnost této zprávy vyprší.
Uzamčení tokenu	ServiceBusMessageHeaders#LOCK_TOKEN	Řetězec	Ne	Token zámku pro aktuální zprávu.
Zamčeno do	ServiceBusMessageHeaders#LOCKED_UNTIL	Posun data a času	Ne	Datum a čas, kdy zámek této zprávy vyprší.
Pořadové číslo	ServiceBusMessageHeaders#SEQUENCE_NUMBER	dlouhý	Ne	Jedinečné číslo přiřazené ke zprávě službou Service Bus.
Stát	ServiceBusMessageHeaders#STATE	ServiceBusMessageState	Ne	Stav zprávy může být aktivní, odložený nebo naplánovaný.

Podpora klíče oddílu

Tento starter podporuje dělení oddílů ve službě Service Bus tím, že umožňuje nastavit klíč oddílu a ID relace v hlavičce zprávy. V této části se dozvíte, jak nastavit klíč oddílu pro zprávy.

Doporučeno: Jako klíč hlavičky použijte ServiceBusMessageHeaders.PARTITION_KEY.

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

Nedoporučuje se, ale aktuálně se podporuje: AzureHeaders.PARTITION_KEY jako klíč hlavičky.

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

Poznámka

Pokud jsou v záhlaví zprávy nastavené ServiceBusMessageHeaders.PARTITION_KEY i AzureHeaders.PARTITION_KEY, je preferován ServiceBusMessageHeaders.PARTITION_KEY.

Podpora relací

Tento příklad ukazuje, jak ručně nastavit ID relace zprávy v aplikaci.

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

Poznámka

Pokud je v záhlaví zprávy nastavena hodnota ServiceBusMessageHeaders.SESSION_ID a současně je nastaveno i jiné záhlaví ServiceBusMessageHeaders.PARTITION_KEY, hodnota identifikátoru relace se nakonec použije k přepsání hodnoty klíče oddílu.

Přizpůsobení vlastností klienta služby Service Bus

Vývojáři můžou pomocí AzureServiceClientBuilderCustomizer přizpůsobit vlastnosti klienta služby Service Bus. Následující příklad přizpůsobí vlastnost sessionIdleTimeout v ServiceBusClientBuilder:

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

Vzorky

Další informace najdete v úložišti na GitHubu azure-spring-boot-samples .

Integrace Springu s Azure Storage Queue

Klíčové koncepty

Azure Queue Storage je služba pro ukládání velkého počtu zpráv. Ke zprávům se dostanete odkudkoli na světě prostřednictvím ověřených volání pomocí protokolu HTTP nebo HTTPS. Zpráva ve frontě může mít velikost až 64 KB. Fronta může obsahovat miliony zpráv až do celkového limitu kapacity účtu úložiště. Fronty se běžně používají k vytvoření zásoby nevyřízené práce, která se pak zpracovává asynchronně.

Nastavení závislostí

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

Konfigurace

Tato úvodní sada nabízí následující možnosti konfigurace:

Vlastnosti konfigurace připojení

Tato část obsahuje možnosti konfigurace pro připojení ke službě Azure Storage Queue.

Poznámka

Pokud se rozhodnete použít objekt zabezpečení k ověření a autorizaci pomocí Microsoft Entra ID pro přístup k prostředku Azure, přečtěte si téma Autorizace přístupu s ID Microsoft Entra, abyste měli jistotu, že objekt zabezpečení má dostatečná oprávnění pro přístup k prostředku Azure.

Konfigurovatelné vlastnosti spring-cloud-azure-starter-integration-storage-queuepřipojení:

Vlastnost	Typ	Popis
spring.cloud.azure.storage.queue.enabled	booleovský	Určuje, jestli je povolená fronta služby Azure Storage.
spring.cloud.azure.storage.queue.connection-string	Řetězec	Hodnota připojovacího řetězce fronty služby Storage
spring.cloud.azure.storage.queue.accountName	Řetězec	Název účtu služby Storage Queue
spring.cloud.azure.storage.queue.accountKey	Řetězec	Klíč účtu fronty úložiště.
spring.cloud.azure.storage.queue.endpoint	Řetězec	Koncový bod služby front úložiště
spring.cloud.azure.storage.queue.sasToken	Řetězec	Přihlašovací údaje tokenu Sas
spring.cloud.azure.storage.queue.serviceVersion	Verze služby QueueService	QueueServiceVersion, který se používá při vytváření požadavků rozhraní API.
spring.cloud.azure.storage.queue.messageEncoding	Řetězec	Kódování zpráv ve frontě

Základní využití

Odesílání zpráv do fronty služby Azure Storage

1. Vyplňte možnosti konfigurace přihlašovacích údajů.

   • Pro přihlašovací údaje jako připojovací řetězec nakonfigurujte v souboru application.yml následující vlastnosti:

     spring:
       cloud:
         azure:
           storage:
             queue:
               connection-string: ${AZURE_STORAGE_QUEUE_CONNECTION_STRING}
     

     Poznámka

     Microsoft doporučuje používat nejbezpečnější dostupný tok ověřování. Tok ověřování popsaný v tomto postupu, například pro databáze, mezipaměti, zasílání zpráv nebo služby AI, vyžaduje velmi vysoký stupeň důvěryhodnosti v aplikaci a nese rizika, která nejsou přítomna v jiných tocích. Tento postup používejte pouze tehdy, pokud bezpečnější možnosti, jako jsou spravované identity pro připojení bez hesla nebo klíče, nejsou vhodné. V případě místních operací počítačů upřednostňujete identity uživatelů pro připojení bez hesla nebo bez klíčů.

   • U přihlašovacích údajů jako spravovaných identit nakonfigurujte v souboru application.yml následující vlastnosti:

     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}
     

Poznámka

Hodnoty povolené pro tenant-id jsou: common, organizations, consumersnebo ID tenanta. Další informace o těchto hodnotách najdete v části Byl použit nesprávný koncový bod (osobní a organizační účty) v článku Chyba AADSTS50020 – Uživatelský účet od zprostředkovatele identity v tenantovi neexistuje. Informace o převodu aplikace s jedním tenantem najdete v tématu Převod jednoklientských aplikací na víceklienta vMicrosoft Entra ID .

• Pro přihlašovací údaje pro instanční objekt služby nakonfigurujte v souboru application.yml následující vlastnosti:

  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}
  

Poznámka

Hodnoty povolené pro tenant-id jsou: common, organizations, consumersnebo ID tenanta. Další informace o těchto hodnotách najdete v části Byl použit nesprávný koncový bod (osobní účty a účty organizací) v článku Chyba AADSTS50020 – Uživatelský účet od poskytovatele identity v tenantovi neexistuje. Informace o převodu aplikace s jedním tenantem najdete v tématu Převod jednoklientských aplikací na víceklienta vMicrosoft Entra ID .

1. Vytvořte DefaultMessageHandler s beanem StorageQueueTemplate pro odesílání zpráv do služby 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;
       }
   }
   

2. Vytvořte propojení brány zpráv s výše uvedenou obslužnou rutinou zpráv prostřednictvím kanálu zpráv.

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

3. Odesílání zpráv pomocí brány

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

Příjem zpráv z fronty služby Azure Storage

1. Vyplňte možnosti konfigurace přihlašovacích údajů.

2. Vytvořte bean kanálu zpráv jako vstupní kanál.

   class Demo {
       private static final String INPUT_CHANNEL = "input";
   
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   }
   

3. Vytvořte StorageQueueMessageSource s bean StorageQueueTemplate pro přijímání zpráv ze služby 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);
       }
   }
   

4. Vytvořte vazbu příjemce zprávy pomocí StorageQueueMessageSource vytvořeného v posledním kroku prostřednictvím kanálu zprávy, který jsme vytvořili dříve.

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

Vzorky

Další informace najdete v úložišti na GitHubu azure-spring-boot-samples .