Obsługa Spring Integration w Spring Cloud Azure

Rozszerzenie Azure dla Spring Integration udostępnia adaptery Spring Integration dla różnych usług oferowanych przez pakiet Azure SDK for Java. Zapewniamy obsługę integracji spring dla tych usług platformy Azure: Event Hubs, Service Bus, Storage Queue. Poniżej znajduje się lista obsługiwanych adapterów:

• spring-cloud-azure-starter-integration-eventhubs — aby uzyskać więcej informacji, zobacz Spring Integration with Azure Event Hubs
• spring-cloud-azure-starter-integration-servicebus — aby uzyskać więcej informacji, zobacz Spring Integration with Azure Service Bus (Integracja Spring z usługą Azure Service Bus
• spring-cloud-azure-starter-integration-storage-queue — aby uzyskać więcej informacji, zobacz Spring Integration with Azure Storage Queue

Integracja platformy Spring z usługą Azure Event Hubs

Kluczowe pojęcia

Azure Event Hubs to platforma analizy strumieniowej danych Big Data i usługa pozyskiwania oraz przetwarzania zdarzeń. Może odbierać i przetwarzać miliony zdarzeń na sekundę. Dane wysłane do centrum zdarzeń można przekształcać i przechowywać za pomocą dowolnego dostawcy analityki czasu rzeczywistego lub adapterów do przetwarzania wsadowego bądź magazynowania.

Spring Integration umożliwia lekką komunikację opartą na komunikatach w aplikacjach opartych na Springu oraz integrację z systemami zewnętrznymi za pomocą deklaratywnych adapterów. Te adaptery zapewniają wyższy poziom abstrakcji w odniesieniu do obsługi przez platformę Spring zdalnych wywołań, komunikatów i harmonogramowania. Projekt rozszerzenia Spring Integration for Event Hubs udostępnia adaptery i bramy kanałów przychodzących i wychodzących dla usługi Azure Event Hubs.

Uwaga

Interfejsy API obsługujące RxJava zostają usunięte od wersji 4.0.0. Aby uzyskać szczegółowe informacje, zobacz Javadoc.

Grupa odbiorców

Usługa Event Hubs zapewnia podobną obsługę grupy odbiorców jako platformę Apache Kafka, ale z niewielką inną logiką. Podczas gdy platforma Kafka przechowuje wszystkie zatwierdzone przesunięcia w brokerze, musisz przechowywać przesunięcia komunikatów usługi Event Hubs przetwarzanych ręcznie. Zestaw SDK usługi Event Hubs udostępnia funkcję do przechowywania takich przesunięć w usłudze Azure Storage.

Obsługa partycjonowania

Usługa Event Hubs oferuje podobną koncepcję fizycznej partycji jak Kafka. Jednak w przeciwieństwie do automatycznego równoważenia obciążenia platformy Kafka między użytkownikami i partycjami usługa Event Hubs zapewnia rodzaj trybu wyprzedzania. Konto magazynu służy jako dzierżawa do określania, która partycja należy do którego odbiorcy. Po uruchomieniu nowego użytkownika spróbuje ukraść niektóre partycje od najbardziej obciążonych użytkowników w celu osiągnięcia równoważenia obciążenia.

Aby określić strategię równoważenia obciążenia, deweloperzy mogą używać EventHubsContainerProperties dla konfiguracji. Zobacz przykład konfiguracji EventHubsContainerProperties w poniższej sekcji.

Obsługa konsumentów usługi Batch

EventHubsInboundChannelAdapter obsługuje tryb przetwarzania wsadowego. Aby to włączyć, użytkownicy mogą ustawić tryb nasłuchiwania na ListenerMode.BATCH podczas tworzenia instancji EventHubsInboundChannelAdapter. Po włączeniu komunikat, którego ładunkiem jest lista zdarzeń wsadowych, zostanie odebrany i przekazany do kanału podrzędnego. Każdy nagłówek wiadomości jest również przekształcany do postaci listy, której zawartość stanowi odpowiadająca mu wartość nagłówka wyodrębniona z każdego zdarzenia. W przypadku wspólnych nagłówków identyfikatora partycji, checkpointera i właściwości ostatnio umieszczonego w kolejce zdarzenia są one przedstawiane jako pojedyncza wartość dla całej partii, jeśli wszystkie zdarzenia w partii mają tę samą wartość. Aby uzyskać więcej informacji, zobacz sekcję Nagłówki komunikatów usługi Event Hubs.

Uwaga

Nagłówek punktu kontrolnego istnieje tylko wtedy, gdy jest używany ręczny tryb punktu kontrolnego.

Punkt kontrolny odbiorcy wsadowego obsługuje dwa tryby: BATCH i MANUAL. BATCH tryb jest trybem automatycznego tworzenia punktów kontrolnych w celu sprawdzenia całej partii zdarzeń jednocześnie po ich odebraniu. MANUAL tryb polega na określeniu punktów kontrolnych zdarzeń przez użytkowników. Jeśli element Checkpointer jest używany, zostanie przekazany w nagłówku wiadomości, a użytkownicy będą mogli użyć go do zapisywania punktów kontrolnych.

Zasady zużycia wsadowego można określić za pomocą właściwości max-size i max-wait-time, gdzie max-size jest właściwością wymaganą, natomiast max-wait-time jest opcjonalna. Aby określić strategię przetwarzania wsadowego, deweloperzy mogą użyć EventHubsContainerProperties do konfiguracji. Przykład konfiguracji EventHubsContainerProperties znajdziesz w poniższej sekcji.

Konfiguracja zależności

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

Konfiguracja

Ten starter udostępnia następujące 3 części opcji konfiguracji:

Właściwości konfiguracji połączenia

Ta sekcja zawiera opcje konfiguracji używane do nawiązywania połączenia z usługą Azure Event Hubs.

Uwaga

Jeśli zdecydujesz się użyć podmiotu zabezpieczeń do uwierzytelniania i autoryzacji za pomocą identyfikatora Entra firmy Microsoft na potrzeby uzyskiwania dostępu do zasobu platformy Azure, zobacz Autoryzuj dostęp za pomocą identyfikatora Entra firmy Microsoft, aby upewnić się, że podmiot zabezpieczeń otrzymał wystarczające uprawnienia dostępu do zasobu platformy Azure.

Konfigurowalne właściwości połączenia spring-cloud-azure-starter-integration-eventhubs:

Własność	Typ	Opis
spring.cloud.azure.eventhubs.enabled	logiczny	Określa, czy usługa Azure Event Hubs jest włączona.
spring.cloud.azure.eventhubs.connection-string	Struna	Wartość parametrów połączenia dla przestrzeni nazw usługi Event Hubs.
spring.cloud.azure.eventhubs.namespace	Struna	Wartość przestrzeni nazw usługi Event Hubs, stanowiąca prefiks nazwy FQDN. FQDN powinien mieć postać NamespaceName.DomainName
spring.cloud.azure.eventhubs.domain-name	Struna	Nazwa domeny dla wartości przestrzeni nazw usługi Azure Event Hubs.
spring.cloud.azure.eventhubs.custom-endpoint-address	Struna	Niestandardowy adres punktu końcowego.
spring.cloud.azure.eventhubs.shared-connection	logiczny	To, czy bazowe klienty EventProcessorClient i EventHubProducerAsyncClient korzystają z tego samego połączenia. Domyślnie dla każdego tworzonego klienta usługi Event Hubs tworzone i używane jest nowe połączenie.

Właściwości konfiguracji punktu kontrolnego

Ta sekcja zawiera opcje konfiguracji usługi Storage Blobs Service, która jest używana do utrwalania własności partycji i informacji o punkcie kontrolnym.

Uwaga

Od wersji 4.0.0, jeśli właściwość spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists nie zostanie włączona ręcznie, żaden kontener Storage nie zostanie utworzony automatycznie.

Tworzenie punktów kontrolnych dla konfigurowalnych właściwości elementu spring-cloud-azure-starter-integration-eventhubs:

Własność	Typ	Opis
spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists	logiczny	Czy zezwolić na tworzenie kontenerów, jeśli jeszcze nie istnieją?
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name	Struna	Nazwa konta magazynu.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key	Struna	Klucz dostępu do konta magazynu.
spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name	Struna	Nazwa kontenera magazynu.

Typowe opcje konfiguracji zestawu SDK usług Azure mają zastosowanie również do magazynu punktów kontrolnych Storage Blob. Obsługiwane opcje konfiguracji są wprowadzane w konfiguracji platformy Azure Spring Cloudi można je skonfigurować za pomocą ujednoliconego prefiksu spring.cloud.azure. lub prefiksu spring.cloud.azure.eventhubs.processor.checkpoint-store..

Właściwości konfiguracji procesora centrum zdarzeń

EventHubsInboundChannelAdapter używa EventProcessorClient do odbierania komunikatów z centrum zdarzeń, a do konfigurowania ogólnych właściwości elementu EventProcessorClient deweloperzy mogą używać EventHubsContainerProperties. Zobacz poniższą sekcję dotyczącą tego, jak pracować z EventHubsInboundChannelAdapter.

Podstawowe użycie

Wysyłanie komunikatów do usługi Azure Event Hubs

1. Wypełnij opcje konfiguracji poświadczeń.

   • Jeśli poświadczenia są podawane jako parametry połączenia, skonfiguruj następujące właściwości w pliku application.yml:

     spring:
       cloud:
         azure:
           eventhubs:
             connection-string: ${AZURE_EVENT_HUBS_CONNECTION_STRING}
             processor:
               checkpoint-store:
                 container-name: ${CHECKPOINT-CONTAINER}
                 account-name: ${CHECKPOINT-STORAGE-ACCOUNT}
                 account-key: ${CHECKPOINT-ACCESS-KEY}
     

     Uwaga

     Firma Microsoft zaleca korzystanie z najbezpieczniejszego dostępnego przepływu uwierzytelniania. Przepływ uwierzytelniania opisany w tej procedurze, taki jak bazy danych, pamięci podręczne, komunikaty lub usługi sztucznej inteligencji, wymaga bardzo wysokiego stopnia zaufania w aplikacji i niesie ze sobą ryzyko, które nie występują w innych przepływach. Użyj tego przepływu tylko wtedy, gdy bardziej bezpieczne opcje, takie jak tożsamości zarządzane dla połączeń bez hasła lub bez kluczy, nie są opłacalne. W przypadku operacji maszyny lokalnej preferuj tożsamości użytkowników dla połączeń bez hasła lub bez klucza.

   • W przypadku poświadczeń jako tożsamości zarządzanych skonfiguruj następujące właściwości w pliku application.yml:

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

   • W przypadku poświadczeń jednostki usługi skonfiguruj następujące właściwości w pliku application.yml:

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

Uwaga

Dozwolone wartości dla tenant-id to: common, organizations, consumerslub identyfikator dzierżawy. Aby uzyskać więcej informacji na temat tych wartości, zobacz sekcję Użyto niewłaściwego punktu końcowego (konta osobiste i organizacyjne) w artykule Błąd AADSTS50020 — konto użytkownika od dostawcy tożsamości nie istnieje w dzierżawie. Aby uzyskać informacje na temat konwersji aplikacji z jedną dzierżawą na aplikację wielodzierżawną, zobacz Konwertowanie aplikacji z jedną dzierżawą na aplikację wielodzierżawną w usłudze Microsoft Entra ID.

1. Utwórz DefaultMessageHandler przy użyciu komponentu bean EventHubsTemplate, aby wysyłać komunikaty do usługi 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. Utwórz powiązanie bramy komunikatów z powyższym programem obsługi komunikatów za pośrednictwem kanału komunikatów.

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

3. Wysyłaj komunikaty przy użyciu bramy.

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

Odbieranie komunikatów z usługi Azure Event Hubs

1. Wypełnij opcje konfiguracji poświadczeń.

2. Utwórz fasolę kanału komunikatów jako kanał wejściowy.

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

3. Utwórz EventHubsInboundChannelAdapter za pomocą beana EventHubsMessageListenerContainer, aby odbierać komunikaty z usługi 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. Utwórz powiązanie odbiornika komunikatów za pomocą elementu EventHubsInboundChannelAdapter przy użyciu wcześniej utworzonego kanału komunikatów.

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

Skonfiguruj EventHubsMessageConverter, aby dostosować objectMapper

EventHubsMessageConverter został utworzony jako konfigurowalny komponent, aby umożliwić użytkownikom dostosowanie ObjectMapper.

Obsługa konsumentów usługi Batch

Aby korzystać z komunikatów z usługi Event Hubs w partiach, podobnie jak w przypadku powyższego przykładu, oprócz tego użytkownicy powinni ustawić opcje konfiguracji związane z przetwarzaniem wsadowym dla EventHubsInboundChannelAdapter.

Podczas tworzenia EventHubsInboundChannelAdapter tryb nasłuchiwania powinien być ustawiony na BATCH. Podczas tworzenia beana EventHubsMessageListenerContainer, ustaw tryb punktu kontrolnego na MANUAL lub BATCH, a opcje przetwarzania wsadowego można skonfigurować zgodnie z potrzebami.

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

Nagłówki komunikatów usługi Event Hubs

W poniższej tabeli pokazano, jak właściwości komunikatów usługi Event Hubs są mapowane na nagłówki komunikatów Spring. W przypadku usługi Azure Event Hubs komunikat jest wywoływany jako event.

Mapowanie między właściwościami komunikatów/zdarzeń w usłudze Event Hubs a nagłówkami komunikatów Spring w trybie odbiornika rekordów:

Właściwości zdarzeń usługi Event Hubs	Stałe nagłówka komunikatów spring	Typ	Opis
Czas w kolejce	EventHubsHeaders#ENQUEUED_TIME	Natychmiast	Chwila, w czasie UTC, umieszczenia zdarzenia w kolejce na partycji usługi Event Hub.
Przesunięcie	EventHubsHeaders#OFFSET	Długi	Przesunięcie zdarzenia w momencie jego odebrania ze skojarzonej partycji usługi Event Hub.
Klucz partycji	AzureHeaders#PARTITION_KEY	Struna	Klucz skrótu partycji, jeśli został ustawiony podczas pierwotnej publikacji zdarzenia.
Identyfikator partycji	AzureHeaders#RAW_PARTITION_ID	Struna	Identyfikator partycji usługi Event Hub.
Numer sekwencji	EventHubsHeaders#SEQUENCE_NUMBER	Długi	Numer sekwencyjny przypisany do zdarzenia w momencie umieszczenia go w kolejce w skojarzonej partycji usługi Event Hub.
Właściwości ostatnio umieszczonego w kolejce zdarzenia	EventHubsHeaders#LAST_ENQUEUED_EVENT_PROPERTIES	LastEnqueuedEventProperties	Właściwości ostatniego zdarzenia umieszczonego w kolejce w tej partycji.
NIE	AzureHeaders#CHECKPOINTER	Punkt kontrolny	Nagłówek punktu kontrolnego dla określonego komunikatu.

Użytkownicy mogą analizować nagłówki komunikatów dla powiązanych informacji o każdym zdarzeniu. Aby ustawić nagłówek komunikatu dla zdarzenia, wszystkie dostosowane nagłówki zostaną umieszczone jako właściwość aplikacji zdarzenia, gdzie nagłówek jest ustawiony jako klucz właściwości. Po odebraniu zdarzeń z usługi Event Hubs wszystkie właściwości aplikacji zostaną przekonwertowane na nagłówek komunikatu.

Uwaga

Nie można ręcznie ustawiać nagłówków komunikatu, takich jak klucz partycji, czas umieszczenia w kolejce, offset i numer sekwencyjny.

Po włączeniu trybu konsumenta wsadowego poniżej wymieniono określone nagłówki komunikatów wsadowych, które zawierają listę wartości z każdego pojedynczego zdarzenia usługi Event Hubs.

Mapowanie między komunikatami usługi Event Hubs/ właściwościami zdarzeń i nagłówkami komunikatów spring w trybie odbiornika usługi Batch:

Właściwości zdarzeń usługi Event Hubs	Stałe nagłówka komunikatu Spring Batch	Typ	Opis
Czas w kolejce	EventHubsHeaders#ENQUEUED_TIME	Lista Instant	Lista znaczników czasu w UTC określających moment umieszczenia każdego zdarzenia w kolejce w partycji centrum zdarzeń.
Przesunięcie	EventHubsHeaders#OFFSET	Lista typu Long	Lista przesunięć każdego zdarzenia w momencie jego odebrania z powiązanej partycji usługi Event Hub.
Klucz partycji	AzureHeaders#PARTITION_KEY	Lista łańcuchów	Lista kluczy haszujących partycji, jeśli zostały ustawione podczas pierwotnej publikacji każdego zdarzenia.
Numer sekwencji	EventHubsHeaders#SEQUENCE_NUMBER	Lista długich	Lista numerów sekwencyjnych przypisanych do poszczególnych zdarzeń w momencie umieszczenia ich w kolejce w powiązanej partycji centrum zdarzeń.
Właściwości systemu	EventHubsHeaders#BATCH_CONVERTED_SYSTEM_PROPERTIES	Lista mapy	Lista właściwości systemowych każdego zdarzenia.
Właściwości aplikacji	EventHubsHeaders#BATCH_CONVERTED_APPLICATION_PROPERTIES	Lista mapy	Lista właściwości aplikacji każdego zdarzenia, w którym są umieszczane wszystkie dostosowane nagłówki komunikatów lub właściwości zdarzenia.

Uwaga

Podczas publikowania komunikatów wszystkie powyższe nagłówki wsadowe zostaną usunięte z komunikatów, jeśli istnieją.

Próbki

Więcej informacji można znaleźć w azure-spring-boot-samples repozytorium w serwisie GitHub.

Integracja platformy Spring z usługą Azure Service Bus

Kluczowe pojęcia

Spring Integration umożliwia lekką komunikację opartą na komunikatach w aplikacjach opartych na Springu oraz integrację z systemami zewnętrznymi za pomocą deklaratywnych adapterów.

Projekt rozszerzenia Spring Integration dla usługi Azure Service Bus udostępnia adaptery kanałowe dla komunikacji przychodzącej i wychodzącej dla usługi Azure Service Bus.

Uwaga

Interfejsy API obsługujące CompletableFuture zostały wycofane począwszy od wersji 2.10.0 i zastąpione przez Reactor Core począwszy od wersji 4.0.0. Aby uzyskać szczegółowe informacje, zobacz Javadoc.

Konfiguracja zależności

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

Konfiguracja

Ten starter udostępnia następujące 2 części opcji konfiguracji:

Właściwości konfiguracji połączenia

Ta sekcja zawiera opcje konfiguracji używane do nawiązywania połączenia z usługą Azure Service Bus.

Uwaga

Jeśli zdecydujesz się użyć podmiotu zabezpieczeń do uwierzytelniania i autoryzacji za pomocą identyfikatora Entra firmy Microsoft na potrzeby uzyskiwania dostępu do zasobu platformy Azure, zobacz Autoryzuj dostęp za pomocą identyfikatora Entra firmy Microsoft, aby upewnić się, że podmiot zabezpieczeń otrzymał wystarczające uprawnienia dostępu do zasobu platformy Azure.

Konfigurowalne właściwości połączenia spring-cloud-azure-starter-integration-servicebus:

Własność	Typ	Opis
spring.cloud.azure.servicebus.enabled	logiczny	Określa, czy usługa Azure Service Bus jest włączona.
spring.cloud.azure.servicebus.connection-string	Struna	Wartość parametrów połączenia przestrzeni nazw usługi Service Bus.
spring.cloud.azure.servicebus.custom-endpoint-address	Struna	Niestandardowy adres punktu końcowego do użycia podczas nawiązywania połączenia z usługą Service Bus.
spring.cloud.azure.servicebus.namespace	Struna	Wartość przestrzeni nazw usługi Service Bus, która jest prefiksem nazwy FQDN. FQDN powinien mieć postać NamespaceName.DomainName
spring.cloud.azure.servicebus.domain-name	Struna	Nazwa domeny wartości przestrzeni nazw usługi Azure Service Bus.

Właściwości konfiguracji procesora usługi Service Bus

ServiceBusInboundChannelAdapter używa ServiceBusProcessorClient do odbierania komunikatów, a aby skonfigurować ogólne właściwości elementu ServiceBusProcessorClient, programiści mogą użyć ServiceBusContainerProperties. Zobacz poniższą sekcję dotyczącą tego, jak pracować z ServiceBusInboundChannelAdapter.

Podstawowe użycie

Wysyłanie komunikatów do usługi Azure Service Bus

1. Wypełnij opcje konfiguracji poświadczeń.

   • Jeśli poświadczenia są podawane jako parametry połączenia, skonfiguruj następujące właściwości w pliku application.yml:

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

     Uwaga

     Firma Microsoft zaleca korzystanie z najbezpieczniejszego dostępnego przepływu uwierzytelniania. Przepływ uwierzytelniania opisany w tej procedurze, taki jak bazy danych, pamięci podręczne, komunikaty lub usługi sztucznej inteligencji, wymaga bardzo wysokiego stopnia zaufania w aplikacji i niesie ze sobą ryzyko, które nie występują w innych przepływach. Użyj tego przepływu tylko wtedy, gdy bardziej bezpieczne opcje, takie jak tożsamości zarządzane dla połączeń bez hasła lub bez kluczy, nie są opłacalne. W przypadku operacji maszyny lokalnej preferuj tożsamości użytkowników dla połączeń bez hasła lub bez klucza.

   • W przypadku poświadczeń jako tożsamości zarządzanych skonfiguruj następujące właściwości w pliku application.yml:

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

Uwaga

Dozwolone wartości dla tenant-id to: common, organizations, consumerslub identyfikator dzierżawy. Aby uzyskać więcej informacji na temat tych wartości, zobacz sekcję Użyto niewłaściwego punktu końcowego (konta osobiste i organizacyjne) w artykule Błąd AADSTS50020 — konto użytkownika od dostawcy tożsamości nie istnieje w dzierżawie. Aby uzyskać informacje na temat konwersji aplikacji z jedną dzierżawą na aplikację wielodzierżawną, zobacz Konwertowanie aplikacji z jedną dzierżawą na aplikację wielodzierżawną w usłudze Microsoft Entra ID.

• W przypadku poświadczeń jednostki usługi skonfiguruj następujące właściwości w pliku application.yml:

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

Uwaga

Dozwolone wartości dla tenant-id to: common, organizations, consumerslub identyfikator dzierżawy. Aby uzyskać więcej informacji na temat tych wartości, zobacz sekcję Użyto niewłaściwego punktu końcowego (konta osobiste i organizacyjne) w artykule Błąd AADSTS50020 — konto użytkownika od dostawcy tożsamości nie istnieje w dzierżawie. Aby uzyskać informacje na temat konwersji aplikacji z jedną dzierżawą na aplikację wielodzierżawną, zobacz Konwertowanie aplikacji z jedną dzierżawą na aplikację wielodzierżawną w usłudze Microsoft Entra ID.

1. Utwórz DefaultMessageHandler za pomocą beana ServiceBusTemplate, aby wysyłać komunikaty do usługi Service Bus, i ustaw typ encji dla elementu ServiceBusTemplate. W tym przykładzie jako przykład wykorzystano kolejkę usługi 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. Utwórz powiązanie bramy komunikatów z powyższym programem obsługi komunikatów za pośrednictwem kanału komunikatów.

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

3. Wysyłaj komunikaty przy użyciu bramy.

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

Odbieranie komunikatów z usługi Azure Service Bus

1. Wypełnij opcje konfiguracji poświadczeń.

2. Utwórz fasolę kanału komunikatów jako kanał wejściowy.

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

3. Utwórz ServiceBusInboundChannelAdapter przy użyciu komponentu ServiceBusMessageListenerContainer, aby odbierać komunikaty z usługi Service Bus. W tym przykładzie jako przykład wykorzystano kolejkę usługi Service Bus.

   @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. Utwórz powiązanie odbiorcy komunikatów z ServiceBusInboundChannelAdapter za pośrednictwem utworzonego wcześniej kanału wiadomości.

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

Konfigurowanie elementu ServiceBusMessageConverter, aby dostosować objectMapper

ServiceBusMessageConverter jest stosowana jako konfigurowalna fasola umożliwiająca użytkownikom dostosowywanie ObjectMapper.

Nagłówki komunikatów usługi Service Bus

W przypadku niektórych nagłówków usługi Service Bus, które można przypisać do wielu stałych nagłówków Spring, podano priorytet różnych nagłówków Spring.

Mapowanie pomiędzy nagłówkami Service Bus a nagłówkami Spring:

Nagłówki i właściwości komunikatów usługi Service Bus	Stałe nagłówka wiadomości spring	Typ	Konfigurowalne	Opis
Typ zawartości	MessageHeaders#CONTENT_TYPE	Struna	Tak	Deskryptor RFC2045 typu zawartości komunikatu.
Identyfikator korelacji	ServiceBusMessageHeaders#CORRELATION_ID	Struna	Tak	Identyfikator korelacji komunikatu
Identyfikator komunikatu	ServiceBusMessageHeaders#MESSAGE_ID	Struna	Tak	Identyfikator wiadomości; ten nagłówek ma wyższy priorytet niż MessageHeaders#ID.
Identyfikator komunikatu	MessageHeaders#ID	UUID	Tak	Identyfikator wiadomości; ten nagłówek ma niższy priorytet niż ServiceBusMessageHeaders#MESSAGE_ID.
Klucz partycji	ServiceBusMessageHeaders#PARTITION_KEY	Struna	Tak	Klucz partycji do wysyłania komunikatu do jednostki podzielonej na partycje.
Odpowiedz na	MessageHeaders#REPLY_CHANNEL	Struna	Tak	Adres jednostki do wysyłania odpowiedzi.
Odpowiedz na identyfikator sesji	ServiceBusMessageHeaders#REPLY_TO_SESSION_ID	Struna	Tak	Wartość właściwości ReplyToGroupId komunikatu.
Zaplanowany czas umieszczenia w kolejce UTC	ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME	OffsetDateTime	Tak	Data i godzina, o której komunikat powinien zostać umieszczony w kolejce w usłudze Service Bus; ten nagłówek ma wyższy priorytet niż AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE.
Zaplanowana godzina kolejki utc	AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE	Liczba całkowita	Tak	Data i godzina, o której komunikat powinien zostać umieszczony w kolejce w usłudze Service Bus; ten nagłówek ma niższy priorytet niż ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME.
Identyfikator sesji	ServiceBusMessageHeaders#SESSION_ID	Struna	Tak	Identyfikator sesji dla jednostki obsługującej sesję.
Czas życia	ServiceBusMessageHeaders#TIME_TO_LIVE	Czas trwania	Tak	Czas trwania przed wygaśnięciem tej wiadomości.
Do	ServiceBusMessageHeaders#TO	Struna	Tak	Adres "do" komunikatu, zarezerwowany do użycia w przyszłości w scenariuszach routingu i obecnie ignorowany przez samego brokera.
Temat	ServiceBusMessageHeaders#SUBJECT	Struna	Tak	Temat wiadomości.
Opis błędu martwych komunikatów	ServiceBusMessageHeaders#DEAD_LETTER_ERROR_DESCRIPTION	Struna	Nie	Opis komunikatu, który został utracony.
Przyczyna utraconych listów	ServiceBusMessageHeaders#DEAD_LETTER_REASON	Struna	Nie	Powodem, dla którego wiadomość została martwa.
Źródło utraconych listów	ServiceBusMessageHeaders#DEAD_LETTER_SOURCE	Struna	Nie	Jednostka, w której wiadomość została martwa.
Liczba dostaw	ServiceBusMessageHeaders#DELIVERY_COUNT	długi	Nie	Liczba przypadków dostarczenia tego komunikatu do klientów.
Numer sekwencji umieszczonej w kolejce	ServiceBusMessageHeaders#ENQUEUED_SEQUENCE_NUMBER	długi	Nie	Numer sekwencyjny nadany komunikatowi po umieszczeniu go w kolejce przez usługę Service Bus.
Czas w kolejce	ServiceBusMessageHeaders#ENQUEUED_TIME	OffsetDateTime	Nie	Data i godzina, o której ten komunikat został umieszczony w kolejce w usłudze Service Bus.
Wygasa o	ServiceBusMessageHeaders#EXPIRES_AT	OffsetDateTime	Nie	Data/godzina wygaśnięcia tej wiadomości.
Blokowanie tokenu	ServiceBusMessageHeaders#LOCK_TOKEN	Struna	Nie	Token blokady dla bieżącego komunikatu.
Zablokowano do	ServiceBusMessageHeaders#LOCKED_UNTIL	OffsetDateTime	Nie	Data/godzina wygaśnięcia blokady tej wiadomości.
Numer sekwencji	ServiceBusMessageHeaders#SEQUENCE_NUMBER	długi	Nie	Unikatowy numer przypisany do komunikatu przez usługę Service Bus.
Stan	ServiceBusMessageHeaders#STATE	ServiceBusMessageState	Nie	Stan komunikatu, który może być aktywny, odroczony lub zaplanowany.

Obsługa klucza partycji

Ten starter obsługuje partycjonowanie usługi Service Bus, umożliwiając ustawienie klucza partycji i identyfikatora sesji w nagłówku komunikatu. W tej sekcji przedstawiono sposób ustawiania klucza partycji dla komunikatów.

Zalecane: użyj ServiceBusMessageHeaders.PARTITION_KEY jako klucza nagłówka.

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

Niezalecane, ale obecnie obsługiwane: AzureHeaders.PARTITION_KEY jako klucz nagłówka.

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

Uwaga

Jeśli obie ServiceBusMessageHeaders.PARTITION_KEY i AzureHeaders.PARTITION_KEY są ustawione w nagłówkach komunikatów, preferowane jest ServiceBusMessageHeaders.PARTITION_KEY.

Obsługa sesji

W tym przykładzie pokazano, jak ręcznie ustawić identyfikator sesji komunikatu w aplikacji.

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

Uwaga

Gdy ServiceBusMessageHeaders.SESSION_ID jest ustawiona w nagłówkach wiadomości, a inny nagłówek ServiceBusMessageHeaders.PARTITION_KEY jest również ustawiony, wartość identyfikatora sesji zostanie ostatecznie użyta do zastąpienia wartości klucza partycji.

Dostosowywanie właściwości klienta usługi Service Bus

Deweloperzy mogą używać AzureServiceClientBuilderCustomizer do dostosowywania właściwości klienta usługi Service Bus. Poniższy przykład dostosowuje właściwość sessionIdleTimeout w ServiceBusClientBuilder:

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

Próbki

Więcej informacji można znaleźć w azure-spring-boot-samples repozytorium w serwisie GitHub.

Integracja platformy Spring z kolejką usługi Azure Storage

Kluczowe pojęcia

Azure Queue Storage to usługa do przechowywania dużej liczby komunikatów. Uzyskujesz dostęp do komunikatów z dowolnego miejsca na świecie za pośrednictwem uwierzytelnionych wywołań przy użyciu protokołu HTTP lub HTTPS. Komunikat kolejki może mieć rozmiar do 64 KB. Kolejka może zawierać miliony wiadomości, aż do całkowitego limitu pojemności konta magazynowego. Kolejki są często używane do tworzenia listy prac w celu asynchronicznego przetwarzania.

Konfiguracja zależności

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

Konfiguracja

Ten starter udostępnia następujące opcje konfiguracji:

Właściwości konfiguracji połączenia

Ta sekcja zawiera opcje konfiguracji używane do nawiązywania połączenia z kolejką usługi Azure Storage.

Uwaga

Jeśli zdecydujesz się użyć podmiotu zabezpieczeń do uwierzytelniania i autoryzacji za pomocą identyfikatora Entra firmy Microsoft na potrzeby uzyskiwania dostępu do zasobu platformy Azure, zobacz Autoryzuj dostęp za pomocą identyfikatora Entra firmy Microsoft, aby upewnić się, że podmiot zabezpieczeń otrzymał wystarczające uprawnienia dostępu do zasobu platformy Azure.

Konfigurowalne właściwości połączenia spring-cloud-azure-starter-integration-storage-queue:

Własność	Typ	Opis
spring.cloud.azure.storage.queue.enabled	logiczny	Określa, czy kolejka usługi Azure Storage jest włączona.
spring.cloud.azure.storage.queue.connection-string	Struna	Wartość ciągu połączenia dla przestrzeni nazw usługi Queue Storage.
spring.cloud.azure.storage.queue.accountName	Struna	Nazwa konta usługi Storage Queue.
spring.cloud.azure.storage.queue.accountKey	Struna	Klucz konta usługi Storage Queue.
spring.cloud.azure.storage.queue.endpoint	Struna	Punkt końcowy usługi kolejki magazynu.
spring.cloud.azure.storage.queue.sasToken	Struna	Poświadczenie tokenu SAS
spring.cloud.azure.storage.queue.serviceVersion	QueueServiceVersion	QueueServiceVersion, która jest używana podczas wykonywania żądań API.
spring.cloud.azure.storage.queue.messageEncoding	Struna	Kodowanie komunikatu kolejki.

Podstawowe użycie

Wysyłanie komunikatów do kolejki usługi Azure Storage

1. Wypełnij opcje konfiguracji poświadczeń.

   • Jeśli poświadczenia są podawane jako parametry połączenia, skonfiguruj następujące właściwości w pliku application.yml:

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

     Uwaga

     Firma Microsoft zaleca korzystanie z najbezpieczniejszego dostępnego przepływu uwierzytelniania. Przepływ uwierzytelniania opisany w tej procedurze, taki jak bazy danych, pamięci podręczne, komunikaty lub usługi sztucznej inteligencji, wymaga bardzo wysokiego stopnia zaufania w aplikacji i niesie ze sobą ryzyko, które nie występują w innych przepływach. Użyj tego przepływu tylko wtedy, gdy bardziej bezpieczne opcje, takie jak tożsamości zarządzane dla połączeń bez hasła lub bez kluczy, nie są opłacalne. W przypadku operacji maszyny lokalnej preferuj tożsamości użytkowników dla połączeń bez hasła lub bez klucza.

   • W przypadku poświadczeń jako tożsamości zarządzanych skonfiguruj następujące właściwości w pliku application.yml:

     spring:
       cloud:
         azure:
           credential:
             managed-identity-enabled: true
             client-id: ${AZURE_CLIENT_ID}
           profile:
             tenant-id: <tenant>
           storage:
             queue:
               account-name: ${AZURE_STORAGE_QUEUE_ACCOUNT_NAME}
     

Uwaga

Dozwolone wartości dla tenant-id to: common, organizations, consumerslub identyfikator dzierżawy. Aby uzyskać więcej informacji na temat tych wartości, zobacz sekcję Użyto niewłaściwego punktu końcowego (konta osobiste i organizacyjne) w artykule Błąd AADSTS50020 — konto użytkownika od dostawcy tożsamości nie istnieje w dzierżawie. Aby uzyskać informacje na temat konwersji aplikacji z jedną dzierżawą na aplikację wielodzierżawną, zobacz Konwertowanie aplikacji z jedną dzierżawą na aplikację wielodzierżawną w usłudze Microsoft Entra ID.

• W przypadku poświadczeń jednostki usługi skonfiguruj następujące właściwości w pliku application.yml:

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

Uwaga

Dozwolone wartości dla tenant-id to: common, organizations, consumerslub identyfikator dzierżawy. Aby uzyskać więcej informacji na temat tych wartości, zobacz sekcję Użyto niewłaściwego punktu końcowego (konta osobiste i organizacyjne) w artykule Błąd AADSTS50020 — konto użytkownika od dostawcy tożsamości nie istnieje w dzierżawie. Aby uzyskać informacje na temat konwersji aplikacji z jedną dzierżawą na aplikację wielodzierżawną, zobacz Konwertowanie aplikacji z jedną dzierżawą na aplikację wielodzierżawną w usłudze Microsoft Entra ID.

1. Utwórz DefaultMessageHandler przy użyciu beana StorageQueueTemplate, aby wysyłać komunikaty do 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. Utwórz powiązanie bramy komunikatów z powyższym programem obsługi komunikatów za pośrednictwem kanału komunikatów.

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

3. Wysyłaj komunikaty przy użyciu bramy.

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

Odbieranie komunikatów z kolejki usługi Azure Storage

1. Wypełnij opcje konfiguracji poświadczeń.

2. Utwórz fasolę kanału komunikatów jako kanał wejściowy.

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

3. Utwórz StorageQueueMessageSource za pomocą komponentu bean StorageQueueTemplate, aby odbierać komunikaty z usługi 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. Utwórz powiązanie odbiorcy komunikatów za pomocą elementu StorageQueueMessageSource utworzonego w ostatnim kroku za pośrednictwem utworzonego wcześniej kanału komunikatów.

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

Próbki

Więcej informacji można znaleźć w azure-spring-boot-samples repozytorium w serwisie GitHub.