Spring Cloud Azure-Unterstützung für Spring Integration

Spring Integration Extension for Azure stellt Spring Integration Adapter für die verschiedenen Dienste bereit, die vom Azure SDK für Javabereitgestellt werden. Wir bieten Unterstützung für Spring Integration für diese Azure-Dienste: Event Hubs, Service Bus, Storage Queue. Es folgt eine Liste der unterstützten Adapter:

• spring-cloud-azure-starter-integration-eventhubs – weitere Informationen finden Sie unter Spring Integration in Azure Event Hubs
• spring-cloud-azure-starter-integration-servicebus – weitere Informationen finden Sie unter Spring Integration in Azure Service Bus
• spring-cloud-azure-starter-integration-storage-queue – weitere Informationen finden Sie unter Spring Integration in Azure Storage Queue

Spring Integration mit Azure Event Hubs

Schlüsselkonzepte

Azure Event Hubs ist eine Big Data Streaming-Plattform und ein Ereignisaufnahmedienst. Sie kann Millionen von Ereignissen pro Sekunde empfangen und verarbeiten. An einen Event Hub gesendete Daten können mithilfe eines beliebigen Echtzeitanalyseanbieters oder Batch-/Speicheradapters transformiert und gespeichert werden.

Spring Integration ermöglicht einfaches Messaging in Spring-basierten Anwendungen und unterstützt die Integration mit externen Systemen über deklarative Adapter. Diese Adapter bieten eine höhere Abstraktionsebene für Springs Unterstützung von Remoting, Messaging und Aufgabenplanung. Das Spring Integration für Event Hubs Erweiterungsprojekt bietet ein- und ausgehende Kanaladapter und Gateways für Azure Event Hubs.

Anmerkung

RxJava-Support-APIs werden ab Version 4.0.0 gelöscht. Weitere Informationen finden Sie unter Javadoc.

Verbrauchergruppe

Event Hubs bietet ähnlich wie Apache Kafka Unterstützung für Consumergruppen, jedoch mit einer leicht abweichenden Logik. Während Kafka alle zugesicherten Offsets im Broker speichert, müssen Sie Offsets von Event Hubs-Nachrichten speichern, die manuell verarbeitet werden. Event Hubs SDK stellt die Funktion zum Speichern solcher Offsets in Azure Storage bereit.

Partitionierungsunterstützung

Event Hubs bietet ein ähnliches Konzept der physischen Partition wie Kafka. Im Gegensatz zu Kafkas automatischem Neuausgleich zwischen Verbrauchern und Partitionen bietet Event Hubs jedoch eine Art präemptiver Modus. Das Speicherkonto dient als Lease-Mechanismus, um zu bestimmen, welchem Consumer eine Partition gehört. Wenn ein neuer Consumer startet, wird er versuchen, einige Partitionen von den am stärksten ausgelasteten Consumern zu übernehmen, um eine ausgewogene Arbeitslastverteilung zu erreichen.

Um die Lastenausgleichsstrategie anzugeben, können Entwickler EventHubsContainerProperties für die Konfiguration verwenden. Ein Beispiel für die Konfiguration von finden Sie im folgenden Abschnitt.

Batch-Consumerunterstützung

Die EventHubsInboundChannelAdapter unterstützt den Batch-Nutzungsmodus. Um ihn zu aktivieren, können Benutzer den Listenermodus beim Erstellen einer ListenerMode.BATCH Instanz als EventHubsInboundChannelAdapter angeben. Wenn diese Option aktiviert ist, wird eine Nachricht, deren Nutzlast eine Liste von Batchereignissen ist, empfangen und an den nachgeschalteten Kanal übergeben. Jeder Nachrichten-Header wird auch als Liste konvertiert, deren Inhalt der zugehörige Headerwert ist, der aus jedem Ereignis geparst wird. Bei den gemeinsamen Headern für Partitions-ID, Checkpointer und die Eigenschaften der zuletzt in die Warteschlange eingereihten Nachricht werden diese als ein einzelner Wert dargestellt, da das gesamte Ereignisbatch denselben Wert teilt. Weitere Informationen finden Sie im Abschnitt Event Hubs-Nachrichtenkopfzeilen.

Anmerkung

Der Prüfpunktheader ist nur vorhanden, wenn MANUELL Prüfpunktmodus verwendet wird.

Checkpointing für Batch-Consumer unterstützt zwei Modi: BATCH und MANUAL. BATCH Modus ist ein automatischer Prüfpunktmodus, um den gesamten Batch von Ereignissen zusammen zu überwachen, sobald sie empfangen wurden. MANUAL-Modus dient dazu, Prüfpunkte für Benutzerereignisse zu erstellen. Wenn er verwendet wird, wird der Checkpointer im Nachrichtenheader übergeben, und Benutzer können ihn für Checkpointing verwenden.

Die Batchaufwendungsrichtlinie kann durch Eigenschaften von max-size und max-wait-timeangegeben werden, wobei max-size eine erforderliche Eigenschaft ist, während max-wait-time optional ist. Um die Strategie für die Batchnutzung anzugeben, können Entwickler EventHubsContainerProperties für die Konfiguration verwenden. Ein Beispiel dafür, wie konfiguriert wird, finden Sie im folgenden Abschnitt.

Einrichtung von Abhängigkeiten

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

Konfiguration

Dieser Starter umfasst die folgenden 3 Teile von Konfigurationsoptionen:

Verbindungskonfigurationseigenschaften

Dieser Abschnitt enthält die Konfigurationsoptionen, die zum Herstellen einer Verbindung mit Azure Event Hubs verwendet werden.

Anmerkung

Wenn Sie einen Sicherheitsprinzipal zum Authentifizieren und Autorisieren mit Microsoft Entra-ID für den Zugriff auf eine Azure-Ressource verwenden, lesen Sie Autorisieren des Zugriffs mit Microsoft Entra ID, um sicherzustellen, dass dem Sicherheitsprinzipal die ausreichende Berechtigung für den Zugriff auf die Azure-Ressource gewährt wurde.

Konfigurierbare Verbindungseigenschaften von spring-cloud-azure-starter-integration-eventhubs:

Eigentum	Typ	Beschreibung
spring.cloud.azure.eventhubs.enabled	boolesch	Gibt an, ob azure Event Hubs aktiviert ist.
spring.cloud.azure.eventhubs.connection-string	Schnur	Event Hubs Namespace-Verbindungszeichenfolgenwert.
spring.cloud.azure.eventhubs.namespace	Schnur	Wert des Event Hubs-Namespace, der das Präfix des FQDN darstellt. Ein FQDN sollte aus NamespaceName.DomainName bestehen.
spring.cloud.azure.eventhubs.domain-name	Schnur	Domänenname eines Azure Event Hubs-Namespacewerts.
spring.cloud.azure.eventhubs.custom-endpoint-address	Schnur	Benutzerdefinierte Endpunktadresse.
spring.cloud.azure.eventhubs.shared-connection	Boolesch	Gibt an, ob der zugrunde liegende EventProcessorClient und EventHubProducerAsyncClient dieselbe Verbindung verwenden. Standardmäßig wird eine neue Verbindung für jeden erstellten Event Hub-Client erstellt und verwendet.

Prüfpunktkonfigurationseigenschaften

Dieser Abschnitt enthält die Konfigurationsoptionen für den Speicher-Blobs-Dienst, der zum Beibehalten des Partitionsbesitzes und der Prüfpunktinformationen verwendet wird.

Anmerkung

Ab Version 4.0.0 wird, wenn die Eigenschaft von spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists nicht manuell aktiviert ist, kein Speichercontainer automatisch erstellt.

Prüfpunkterstellung für konfigurierbare Eigenschaften von spring-cloud-azure-starter-integration-eventhubs:

Eigentum	Typ	Beschreibung
spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists	Boolesch	Gibt an, ob Container erstellt werden sollen, wenn sie noch nicht vorhanden sind.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name	Schnur	Name für das Speicherkonto.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key	Schnur	Zugriffsschlüssel für Speicherkonto.
spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name	Schnur	Name des Speichercontainers.

Allgemeine Konfigurationsoptionen des Azure Service SDK können ebenfalls für den Prüfpunktspeicher für Storage Blobs konfiguriert werden. Die unterstützten Konfigurationsoptionen werden in Spring Cloud Azure-Konfigurationeingeführt und können entweder mit dem einheitlichen Präfix spring.cloud.azure. oder dem Präfix von spring.cloud.azure.eventhubs.processor.checkpoint-store.konfiguriert werden.

Konfigurationseigenschaften des Event Hub-Prozessors

Die EventHubsInboundChannelAdapter verwendet die EventProcessorClient, um Nachrichten aus einem Event Hub zu verarbeiten. Zum Konfigurieren der allgemeinen Eigenschaften eines EventProcessorClient können Entwickler EventHubsContainerProperties verwenden. Informationen zum Arbeiten mit finden Sie im folgenden Abschnitt.

Grundlegende Nutzung

Senden von Nachrichten an Azure Event Hubs

1. Füllen Sie die Konfigurationsoptionen für die Zugangsdaten aus.

   • Konfigurieren Sie für Anmeldeinformationen als Verbindungszeichenfolge die folgenden Eigenschaften in Ihrer application.yml Datei:

     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}
     

     Anmerkung

     Microsoft empfiehlt, immer den sichersten Authentifizierungsflow zu verwenden, der verfügbar ist. Der in diesem Verfahren beschriebene Authentifizierungsflow, beispielsweise für Datenbanken, Zwischenspeicher, Nachrichten oder KI-Dienste, erfordert ein sehr hohes Maß an Vertrauen in die Anwendung und birgt Risiken, die bei anderen Flows nicht vorhanden sind. Verwenden Sie diesen Flow nur, wenn sicherere Optionen wie verwaltete Identitäten für kennwortlose oder schlüssellose Verbindungen nicht geeignet sind. Bei Vorgängen des lokalen Computers bevorzugen Sie Benutzeridentitäten für kennwortlose oder schlüssellose Verbindungen.

   • Wenn Sie verwaltete Identitäten als Anmeldeinformationen verwenden, konfigurieren Sie die folgenden Eigenschaften in Ihrer application.yml-Datei:

     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}
     

   • Konfigurieren Sie für Anmeldeinformationen als Dienstprinzipal die folgenden Eigenschaften in Ihrer application.yml-Datei:

     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}
     

Anmerkung

Die für tenant-id zulässigen Werte sind: common, organizations, consumersoder die Mandanten-ID. Weitere Informationen zu diesen Werten finden Sie im Abschnitt Falschen Endpunkt verwendet (persönliche Konten und Organisationskonten) in Fehler AADSTS50020 – Benutzerkonto des Identitätsanbieters ist im Mandanten nicht vorhanden. Informationen zum Umwandeln Ihrer Single-Tenant-App finden Sie unter Single-Tenant-App in eine Multitenant-App auf Microsoft Entra ID umwandeln.

1. Erstellen Sie DefaultMessageHandler mit dem EventHubsTemplate-Bean, um Nachrichten an Event Hubs zu senden.

   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. Erstellen Sie über einen Nachrichtenkanal eine Bindung für das Message Gateway mit dem oben genannten Nachrichten-Handler.

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

3. Senden von Nachrichten mithilfe des Gateways.

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

Empfangen von Nachrichten von Azure Event Hubs

1. Füllen Sie die Konfigurationsoptionen für die Zugangsdaten aus.

2. Erstellen Sie einen Bean des Nachrichtenkanals als Eingabekanal.

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

3. Erstellen Sie EventHubsInboundChannelAdapter mithilfe des Beans EventHubsMessageListenerContainer, um Nachrichten von Event Hubs zu empfangen.

   @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. Erstellen Sie eine Nachrichtenempfängerbindung mit EventHubsInboundChannelAdapter über den zuvor erstellten Nachrichtenkanal.

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

Konfigurieren von EventHubsMessageConverter zum Anpassen von objectMapper

EventHubsMessageConverter wird als konfigurierbare Bohnen erstellt, damit Benutzer ObjectMapper anpassen können.

Batch-Consumerunterstützung

Das stapelweise Verarbeiten von Nachrichten aus Event Hubs ähnelt dem obigen Beispiel; zusätzlich sollten Benutzer die Konfigurationsoptionen für die Batchverarbeitung für EventHubsInboundChannelAdapter festlegen.

Beim Erstellen EventHubsInboundChannelAdaptersollte der Listenermodus als BATCHfestgelegt werden. Wenn Sie die Bean EventHubsMessageListenerContainer erstellen, stellen Sie den Prüfpunktmodus entweder auf MANUAL oder BATCH ein, und konfigurieren Sie die Batch-Optionen bei Bedarf.

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

Event Hubs-Nachrichtenkopfzeilen

In der folgenden Tabelle wird veranschaulicht, wie Die Nachrichteneigenschaften von Event Hubs den Kopfzeilen von Spring-Nachrichten zugeordnet werden. Für Azure Event Hubs wird die Nachricht als eventaufgerufen.

Zuordnung zwischen Nachrichten-/Ereigniseigenschaften in Event Hubs und Spring-Message-Headern im Record-Listener-Modus:

Event Hubs-Ereigniseigenschaften	Spring-Message-Header-Konstanten	Typ	Beschreibung
Zeitpunkt der Einreihung	EventHubsHeaders#ENQUEUED_TIME	Sofort	Der Zeitpunkt (in UTC), zu dem das Ereignis in die Partition des Event Hubs eingereiht wurde.
Offset	EventHubsHeaders#OFFSET	Lang	Der Offset des Ereignisses, als es von der zugeordneten Event Hub-Partition empfangen wurde.
Partitionsschlüssel	AzureHeaders#PARTITION_KEY	Schnur	Der Partitionshashschlüssel, wenn er beim ursprünglichen Veröffentlichen des Ereignisses festgelegt wurde.
Partitions-ID	AzureHeaders#RAW_PARTITION_ID	Schnur	Die Partitions-ID des Event Hub.
Sequenznummer	EventHubsHeaders#SEQUENCE_NUMBER	Lang	Die Sequenznummer, die dem Ereignis zugewiesen wurde, als es in die zugeordnete Event Hub-Partition eingereiht wurde.
Eigenschaften des zuletzt in die Warteschlange eingereihten Ereignisses	EventHubsHeaders#LAST_ENQUEUED_EVENT_PROPERTIES	Eigenschaften des zuletzt in die Warteschlange eingereihten Ereignisses	Die Eigenschaften des letzten enqueued-Ereignisses in dieser Partition.
NA	AzureHeaders#PRÜFPUNKT	Checkpointer	Die Kopfzeile für den Prüfpunkt der jeweiligen Nachricht.

Benutzer können die Nachrichtenkopfzeilen für die zugehörigen Informationen jedes Ereignisses analysieren. Um einen Nachrichtenkopf für das Ereignis festzulegen, werden alle angepassten Kopfzeilen als Anwendungseigenschaft eines Ereignisses platziert, wobei der Header als Eigenschaftsschlüssel festgelegt wird. Wenn Ereignisse von Event Hubs empfangen werden, werden alle Anwendungseigenschaften in den Nachrichtenkopf konvertiert.

Anmerkung

Nachrichtenheader wie Partitionsschlüssel, Einreihungszeit, Offset und Sequenznummer können nicht manuell festgelegt werden.

Wenn der Batch-Consumer-Modus aktiviert ist, werden die spezifischen Kopfzeilen von Batchnachrichten wie folgt aufgelistet, die eine Liste der Werte aus jedem einzelnen Event Hubs-Ereignis enthalten.

Zuordnung zwischen Nachrichten-/Ereigniseigenschaften von Event Hubs und Spring-Nachrichtenheadern im Batch-Listener-Modus:

Event Hubs-Ereigniseigenschaften	Spring Batch-Nachrichtenkopfkonstanten	Typ	Beschreibung
Zeitpunkt der Einreihung	EventHubsHeaders#ENQUEUED_TIME	Liste der Instant Messages	Liste der Zeitpunkte in UTC, zu denen jedes Ereignis in die Event Hub-Partition eingereiht wurde.
Offset	EventHubsHeaders#OFFSET	Liste von Long	Liste der Offsets der einzelnen Ereignisse zum Zeitpunkt ihres Empfangs aus der zugeordneten Event Hub-Partition.
Partitionsschlüssel	AzureHeaders#PARTITION_KEY	Liste der Zeichenfolgen	Liste des Partitionshashingschlüssels, wenn er beim ursprünglichen Veröffentlichen jedes Ereignisses festgelegt wurde.
Sequenznummer	EventHubsHeaders#SEQUENCE_NUMBER	Liste der langen Nachrichten	Liste der Sequenznummer, die jedem Ereignis zugewiesen wurde, als es in der zugeordneten Event Hub-Partition abgefragt wurde.
Systemeigenschaften	EventHubsHeaders#BATCH_CONVERTED_SYSTEM_PROPERTIES	Liste der Karte	Liste der Systemeigenschaften jedes Ereignisses.
Anwendungseigenschaften	EventHubsHeaders#BATCH_CONVERTED_APPLICATION_PROPERTIES	Liste der Karten	Liste der Anwendungseigenschaften jedes Ereignisses, in denen alle angepassten Nachrichtenkopfzeilen oder Ereigniseigenschaften platziert werden.

Anmerkung

Bei der Veröffentlichung von Nachrichten werden alle oben genannten Batchheader aus den Nachrichten entfernt, sofern vorhanden.

Proben

Weitere Informationen finden Sie im azure-spring-boot-samples Repository auf GitHub.

Spring Integration mit Azure Service Bus

Schlüsselkonzepte

Spring Integration ermöglicht einfaches Messaging in Spring-basierten Anwendungen und unterstützt die Integration mit externen Systemen über deklarative Adapter.

Das Spring Integration for Azure Service Bus-Erweiterungsprojekt bietet ein- und ausgehende Kanaladapter für Azure Service Bus.

Anmerkung

CompletableFuture-Support-APIs sind ab Version 2.10.0 veraltet und werden von Reaktorkern ab Version 4.0.0 ersetzt. Weitere Informationen finden Sie unter Javadoc.

Einrichtung von Abhängigkeiten

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

Konfiguration

Dieser Start bietet die folgenden 2 Teile der Konfigurationsoptionen:

Verbindungskonfigurationseigenschaften

Dieser Abschnitt enthält die Konfigurationsoptionen, die zum Herstellen einer Verbindung mit Azure Service Bus verwendet werden.

Anmerkung

Wenn Sie einen Sicherheitsprinzipal zum Authentifizieren und Autorisieren mit Microsoft Entra-ID für den Zugriff auf eine Azure-Ressource verwenden, lesen Sie Autorisieren des Zugriffs mit Microsoft Entra ID, um sicherzustellen, dass dem Sicherheitsprinzipal die ausreichende Berechtigung für den Zugriff auf die Azure-Ressource gewährt wurde.

Konfigurierbare Verbindungseigenschaften von spring-cloud-azure-starter-integration-servicebus:

Eigentum	Typ	Beschreibung
spring.cloud.azure.servicebus.enabled	boolesch	Gibt an, ob ein Azure Service Bus aktiviert ist.
spring.cloud.azure.servicebus.connection-string	Schnur	Service Bus-Namespace-Verbindungszeichenfolgenwert.
spring.cloud.azure.servicebus.custom-endpoint-address	Schnur	Die benutzerdefinierte Endpunktadresse, die beim Herstellen einer Verbindung mit Service Bus verwendet werden soll.
spring.cloud.azure.servicebus.namespace	Schnur	Wert des Service Bus-Namespace, der das Präfix der FQDN ist. Ein FQDN sollte aus NamespaceName.DomainName bestehen.
spring.cloud.azure.servicebus.domain-name	Schnur	Domänenname eines Azure Service Bus-Namespacewerts.

Konfigurationseigenschaften des Dienstbusprozessors

Die ServiceBusInboundChannelAdapter verwendet die ServiceBusProcessorClient, um Nachrichten zu verarbeiten. Um die allgemeinen Eigenschaften eines ServiceBusProcessorClient zu konfigurieren, können Entwickler ServiceBusContainerProperties für die Konfiguration verwenden. Informationen zum Arbeiten mit finden Sie im folgenden Abschnitt.

Grundlegende Nutzung

Senden von Nachrichten an Azure Service Bus

1. Füllen Sie die Konfigurationsoptionen für die Zugangsdaten aus.

   • Konfigurieren Sie für Anmeldeinformationen als Verbindungszeichenfolge die folgenden Eigenschaften in Ihrer application.yml Datei:

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

     Anmerkung

     Microsoft empfiehlt, immer den sichersten Authentifizierungsflow zu verwenden, der verfügbar ist. Der in diesem Verfahren beschriebene Authentifizierungsflow, beispielsweise für Datenbanken, Zwischenspeicher, Nachrichten oder KI-Dienste, erfordert ein sehr hohes Maß an Vertrauen in die Anwendung und birgt Risiken, die bei anderen Flows nicht vorhanden sind. Verwenden Sie diesen Flow nur, wenn sicherere Optionen wie verwaltete Identitäten für kennwortlose oder schlüssellose Verbindungen nicht geeignet sind. Bei Vorgängen des lokalen Computers bevorzugen Sie Benutzeridentitäten für kennwortlose oder schlüssellose Verbindungen.

   • Wenn Sie verwaltete Identitäten als Anmeldeinformationen verwenden, konfigurieren Sie die folgenden Eigenschaften in der Datei 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}
     

Anmerkung

Die für tenant-id zulässigen Werte sind: common, organizations, consumersoder die Mandanten-ID. Weitere Informationen zu diesen Werten finden Sie im Abschnitt Falschen Endpunkt verwendet (persönliche und Organisationskonten) von Fehler AADSTS50020 – Benutzerkonto des Identitätsanbieters ist im Mandanten nicht vorhanden. Informationen zum Konvertieren Ihrer Single-Tenant-App finden Sie unter Single-Tenant-App in Microsoft Entra ID in eine Mehrmandanten-App konvertieren.

• Konfigurieren Sie für Anmeldeinformationen als Dienstprinzipal die folgenden Eigenschaften in Ihrer application.yml-Datei:

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

Anmerkung

Die für tenant-id zulässigen Werte sind: common, organizations, consumersoder die Mandanten-ID. Weitere Informationen zu diesen Werten finden Sie im Abschnitt Falschen Endpunkt verwendet (persönliche Konten und Organisationskonten) von Fehler AADSTS50020 – Das Benutzerkonto des Identitätsanbieters ist im Mandanten nicht vorhanden. Informationen zum Konvertieren Ihrer Single-Tenant-App finden Sie unter Single-Tenant-App in eine App für mehrere Mandanten in Microsoft Entra ID konvertieren.

1. Erstellen Sie DefaultMessageHandler mit dem ServiceBusTemplate-Bean, um Nachrichten an den Service Bus zu senden, und legen Sie den Entitätstyp für das ServiceBusTemplate fest. Dieses Beispiel verwendet eine Service Bus-Warteschlange als Beispiel.

   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. Erstellen Sie über einen Nachrichtenkanal eine Bindung für das Message Gateway mit dem oben genannten Nachrichten-Handler.

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

3. Senden von Nachrichten mithilfe des Gateways.

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

Empfangen von Nachrichten von Azure Service Bus

1. Füllen Sie die Konfigurationsoptionen für die Zugangsdaten aus.

2. Erstellen Sie einen Bean des Nachrichtenkanals als Eingabekanal.

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

3. Erstellen Sie ServiceBusInboundChannelAdapter mit dem ServiceBusMessageListenerContainer Bean, um Nachrichten an Service Bus zu empfangen. Dieses Beispiel verwendet eine Service Bus-Warteschlange als Beispiel.

   @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. Erstellen Sie eine Nachrichtenempfängerbindung mit ServiceBusInboundChannelAdapter über den zuvor erstellten Nachrichtenkanal.

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

Konfigurieren von ServiceBusMessageConverter zum Anpassen von objectMapper

ServiceBusMessageConverter ist als konfigurierbare Bean konzipiert, damit Benutzer ObjectMapper anpassen können.

Dienstbus-Nachrichtenkopfzeilen

Bei einigen Service Bus-Headern, die mehreren Spring-Header-Konstanten zugeordnet werden können, ist die Priorität der verschiedenen Spring-Header aufgeführt.

Zuordnung zwischen Service Bus-Headern und Spring-Headern:

Service Bus-Nachrichtenkopfzeilen und -eigenschaften	Spring-Nachrichten-Header-Konstanten	Typ	Konfigurierbar	Beschreibung
Inhaltstyp	MessageHeaders#CONTENT_TYPE	Schnur	Ja	Der RFC2045 Inhaltstypdeskriptor der Nachricht.
Korrelations-ID	ServiceBusMessageHeaders#CORRELATION_ID	Schnur	Ja	Die Korrelations-ID der Nachricht
Nachrichten-ID	ServiceBusMessageHeaders#MESSAGE_ID	Schnur	Ja	Die Nachrichten-ID der Nachricht, diese Kopfzeile hat höhere Priorität als MessageHeaders#ID.
Nachrichten-ID	MessageHeaders#ID	UUID	Ja	Die Nachrichten-ID der Nachricht, diese Kopfzeile hat eine niedrigere Priorität als ServiceBusMessageHeaders#MESSAGE_ID.
Partitionsschlüssel	ServiceBusMessageHeaders#PARTITION_KEY	Schnur	Ja	Der Partitionsschlüssel zum Senden der Nachricht an eine partitionierte Entität.
Antwort an	MessageHeaders#REPLY_CHANNEL	Schnur	Ja	Die Adresse einer Entität, an die Antworten gesendet werden sollen.
Antwort auf Sitzungs-ID	ServiceBusMessageHeaders#REPLY_TO_SESSION_ID	Schnur	Ja	Der Wert der ReplyToGroupId-Eigenschaft der Nachricht.
Geplanter Zeitpunkt der Einreihung in die Warteschlange (UTC)	ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME	OffsetDateTime	Ja	Das Datum und die Uhrzeit, zu dem bzw. zu der die Nachricht in Service Bus in die Warteschlange eingereiht werden soll; dieser Header hat eine höhere Priorität als AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE.
Geplanter Zeitpunkt der Einreihung in die Warteschlange (UTC)	AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE	Ganze Zahl	Ja	Das Datum und die Uhrzeit, zu denen die Nachricht in Service Bus in die Warteschlange eingereiht werden soll; dieser Header hat eine niedrigere Priorität als ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME.
Sitzungs-ID	ServiceBusMessageHeaders#SESSION_ID	Schnur	Ja	Die Sitzungskennung für eine sitzungsbewusste Entität.
Zeit zum Leben	ServiceBusMessageHeaders#TIME_TO_LIVE	Dauer	Ja	Die Dauer, bis diese Nachricht abläuft.
An	ServiceBusMessageHeaders#TO	Schnur	Ja	Die "to"-Adresse der Nachricht, die für die zukünftige Verwendung in Routingszenarien reserviert ist und derzeit vom Broker selbst ignoriert wird.
Betreff	ServiceBusMessageHeaders#SUBJECT	Schnur	Ja	Der Betreff für die Nachricht.
Beschreibung des Dead-Letter-Fehlers	ServiceBusMessageHeaders#DEAD_LETTER_ERROR_DESCRIPTION	Schnur	Nein	Die Beschreibung für eine Nachricht, die in die Dead-Letter-Warteschlange verschoben wurde.
Ursache für unzustellbare Nachrichten	ServiceBusMessageHeaders#DEAD_LETTER_REASON	Schnur	Nein	Der Grund, warum eine Nachricht in die Dead-Letter-Warteschlange verschoben wurde.
Quelle von unzustellbaren Nachrichten	ServiceBusMessageHeaders#DEAD_LETTER_SOURCE	Schnur	Nein	Die Entität, in der die Nachricht unzustellbar war.
Lieferanzahl	ServiceBusMessageHeaders#DELIVERY_COUNT	lang	Nein	Die Häufigkeit, mit der diese Nachricht an Clients übermittelt wurde.
Sequenznummer in der Warteschlange	ServiceBusMessageHeaders#ENQUEUED_SEQUENCE_NUMBER	lang	Nein	Die queuierte Sequenznummer, die einer Nachricht von Service Bus zugewiesen ist.
Zeitpunkt der Einreihung	ServiceBusMessageHeaders#ENQUEUED_TIME	OffsetDateTime	Nein	Das Datum, zu dem diese Nachricht in Service Bus enqueuiert wurde.
Läuft ab am	ServiceBusMessageHeaders#EXPIRES_AT	OffsetDateTime	Nein	Das Datum, zu dem diese Nachricht abläuft.
Sperrtoken	ServiceBusMessageHeaders#LOCK_TOKEN	Schnur	Nein	Das Sperrtoken für die aktuelle Nachricht.
Gesperrt bis	ServiceBusMessageHeaders#LOCKED_UNTIL	OffsetDateTime	Nein	Das Datum, zu dem die Sperre dieser Nachricht abläuft.
Sequenznummer	ServiceBusMessageHeaders#SEQUENCE_NUMBER	lang	Nein	Die eindeutige Nummer, die einer Nachricht von Service Bus zugewiesen ist.
Zustand	ServiceBusMessageHeaders#STATE	ServiceBusMessageState	Nein	Der Status der Nachricht, die aktiv, verzögert oder geplant sein kann.

Unterstützung für Partitionsschlüssel

Dieser Starter unterstützt Service Bus-Partitionierung durch Festlegen des Partitionsschlüssels und der Sitzungs-ID im Nachrichtenheader. In diesem Abschnitt wird erläutert, wie Sie den Partitionsschlüssel für Nachrichten festlegen.

Empfohlen: Verwenden Sie ServiceBusMessageHeaders.PARTITION_KEY als Schlüssel der Kopfzeile.

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

Nicht empfohlen, derzeit unterstützt: AzureHeaders.PARTITION_KEY als Schlüssel der Kopfzeile.

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

Anmerkung

Wenn sowohl ServiceBusMessageHeaders.PARTITION_KEY als auch AzureHeaders.PARTITION_KEY in den Nachrichtenkopfzeilen festgelegt werden, wird ServiceBusMessageHeaders.PARTITION_KEY bevorzugt.

Sitzungsunterstützung

In diesem Beispiel wird veranschaulicht, wie Die Sitzungs-ID einer Nachricht in der Anwendung manuell festgelegt wird.

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

Anmerkung

Wenn die ServiceBusMessageHeaders.SESSION_ID in den Nachrichtenkopfzeilen festgelegt wird und auch ein anderer ServiceBusMessageHeaders.PARTITION_KEY Header festgelegt wird, wird der Wert der Sitzungs-ID schließlich verwendet, um den Wert des Partitionsschlüssels zu überschreiben.

Anpassen von ServiceBus-Clienteigenschaften

Entwickler können AzureServiceClientBuilderCustomizer zum Anpassen von Service Bus-Clienteigenschaften verwenden. Im folgenden Beispiel wird die eigenschaft sessionIdleTimeout in ServiceBusClientBuilderangepasst:

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

Proben

Weitere Informationen finden Sie im azure-spring-boot-samples Repository auf GitHub.

Spring Integration mit Azure-Speicherwarteschlangen

Schlüsselkonzepte

Azure Queue Storage ist ein Dienst zum Speichern großer Anzahl von Nachrichten. Sie greifen über authentifizierte Anrufe über HTTP oder HTTPS auf Nachrichten von überall auf der Welt zu. Eine Warteschlangennachricht kann bis zu 64 KB groß sein. Eine Warteschlange kann Millionen von Nachrichten bis zum Gesamtkapazitätslimit eines Speicherkontos enthalten. Warteschlangen werden häufig verwendet, um einen Arbeitsbestand zu erstellen, der asynchron verarbeitet werden kann.

Einrichtung von Abhängigkeiten

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

Konfiguration

Dieser Start bietet die folgenden Konfigurationsoptionen:

Verbindungskonfigurationseigenschaften

Dieser Abschnitt enthält die Konfigurationsoptionen, die zum Herstellen einer Verbindung mit der Azure Storage-Warteschlange verwendet werden.

Anmerkung

Wenn Sie einen Sicherheitsprinzipal zum Authentifizieren und Autorisieren mit Microsoft Entra-ID für den Zugriff auf eine Azure-Ressource verwenden, lesen Sie Autorisieren des Zugriffs mit Microsoft Entra ID, um sicherzustellen, dass dem Sicherheitsprinzipal die ausreichende Berechtigung für den Zugriff auf die Azure-Ressource gewährt wurde.

Konfigurierbare Verbindungseigenschaften von spring-cloud-azure-starter-integration-storage-queue:

Eigentum	Typ	Beschreibung
spring.cloud.azure.storage.queue.enabled	boolesch	Gibt an, ob eine Azure Storage-Warteschlange aktiviert ist.
spring.cloud.azure.storage.queue.connection-string	Schnur	Wert der Verbindungszeichenfolge für den Namespace der Speicherwarteschlange.
spring.cloud.azure.storage.queue.accountName	Schnur	Name des Warteschlangenspeicherkontos.
spring.cloud.azure.storage.queue.accountKey	Schnur	Kontoschlüssel der Storage Queue.
spring.cloud.azure.storage.queue.endpoint	Schnur	Endpunkt des Storage Queue-Dienstes.
spring.cloud.azure.storage.queue.sasToken	Schnur	Sas-Token-Anmeldeinformationen
spring.cloud.azure.storage.queue.serviceVersion	QueueServiceVersion	QueueServiceVersion, die beim Erstellen von API-Anforderungen verwendet wird.
spring.cloud.azure.storage.queue.messageEncoding	Schnur	Codierung von Warteschlangennachrichten.

Grundlegende Nutzung

Senden von Nachrichten an die Azure Storage-Warteschlange

1. Füllen Sie die Konfigurationsoptionen für die Zugangsdaten aus.

   • Konfigurieren Sie für Anmeldeinformationen als Verbindungszeichenfolge die folgenden Eigenschaften in Ihrer application.yml Datei:

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

     Anmerkung

     Microsoft empfiehlt, immer den sichersten Authentifizierungsflow zu verwenden, der verfügbar ist. Der in diesem Verfahren beschriebene Authentifizierungsflow, beispielsweise für Datenbanken, Zwischenspeicher, Nachrichten oder KI-Dienste, erfordert ein sehr hohes Maß an Vertrauen in die Anwendung und birgt Risiken, die bei anderen Flows nicht vorhanden sind. Verwenden Sie diesen Flow nur, wenn sicherere Optionen wie verwaltete Identitäten für kennwortlose oder schlüssellose Verbindungen nicht geeignet sind. Bei Vorgängen des lokalen Computers bevorzugen Sie Benutzeridentitäten für kennwortlose oder schlüssellose Verbindungen.

   • Wenn Sie verwaltete Identitäten als Anmeldeinformationen verwenden, konfigurieren Sie die folgenden Eigenschaften in Ihrer application.yml-Datei:

     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}
     

Anmerkung

Die für tenant-id zulässigen Werte sind: common, organizations, consumersoder die Mandanten-ID. Weitere Informationen zu diesen Werten finden Sie im Abschnitt Falschen Endpunkt verwendet (persönliche Konten und Organisationskonten) in Fehler AADSTS50020 – Benutzerkonto des Identitätsanbieters ist im Mandanten nicht vorhanden. Informationen zum Umwandeln Ihrer Single-Tenant-App finden Sie unter Single-Tenant-App in eine Multitenant-App auf Microsoft Entra ID umwandeln.

• Konfigurieren Sie für Anmeldeinformationen als Dienstprinzipal die folgenden Eigenschaften in Ihrer application.yml-Datei:

  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}
  

Anmerkung

Die für tenant-id zulässigen Werte sind: common, organizations, consumersoder die Mandanten-ID. Weitere Informationen zu diesen Werten finden Sie im Abschnitt Falschen Endpunkt verwendet (persönliche Konten und Organisationskonten) in Fehler AADSTS50020 – Benutzerkonto des Identitätsanbieters ist im Mandanten nicht vorhanden. Informationen zum Umwandeln Ihrer Single-Tenant-App finden Sie unter Single-Tenant-App in eine Multitenant-App auf Microsoft Entra ID umwandeln.

1. Erstellen Sie DefaultMessageHandler mit dem StorageQueueTemplate Bean, um Nachrichten an die Speicherwarteschlange zu senden.

   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. Erstellen Sie über einen Nachrichtenkanal eine Message-Gateway-Bindung mit dem oben genannten Nachrichten-Handler.

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

3. Senden von Nachrichten mithilfe des Gateways.

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

Empfangen von Nachrichten aus der Azure Storage-Warteschlange

1. Füllen Sie die Konfigurationsoptionen für die Zugangsdaten aus.

2. Erstellen Sie einen Bean des Nachrichtenkanals als Eingabekanal.

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

3. Erstellen Sie StorageQueueMessageSource mit dem StorageQueueTemplate bean, um Nachrichten aus Storage Queue zu empfangen.

   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. Erstellen Sie eine Nachrichtenempfängerbindung mit StorageQueueMessageSource, die im letzten Schritt über den zuvor erstellten Nachrichtenkanal erstellt wurde.

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

Proben

Weitere Informationen finden Sie im azure-spring-boot-samples Repository auf GitHub.