Spring Cloud Azure-stöd för Spring Cloud Stream

  • Gäller för: ✅ Version 4.20.0, ✅ Version 5.25.0, ✅ Version 6.5.0, ✅ Version 7.4.0

Spring Cloud Stream är ett ramverk för att skapa mycket skalbara händelsedrivna mikrotjänster som är anslutna till delade meddelandesystem.

Ramverket tillhandahåller en flexibel programmeringsmodell som bygger på redan etablerade och välbekanta Spring-idiom och bästa praxis. Dessa metodtips omfattar stöd för beständiga pub/sub-semantik, konsumentgrupper och tillståndskänsliga partitioner.

Nuvarande binder-implementeringar inkluderar:

Spring Cloud Stream Binder för Azure Event Hubs

Viktiga begrepp

Spring Cloud Stream Binder för Azure Event Hubs tillhandahåller bindningsimplementeringen för Spring Cloud Stream-ramverket. Den här implementeringen använder Spring Integration Event Hubs-kanaladaptrar som grund. Från designens perspektiv liknar Event Hubs Kafka. Event Hubs kan också nås via Kafka API. Om ditt projekt har ett nära beroende av Kafka API kan du prova Events Hub med Kafka API Sample

Konsumentgrupp

Event Hubs har liknande stöd för konsumentgruppen som Apache Kafka, men med lite annan logik. Medan Kafka lagrar alla bekräftade offset i brokern måste du lagra offset för Event Hubs-meddelanden som bearbetas manuellt. Event Hubs SDK gör det möjligt att lagra sådana offsetvärden i Azure Storage.

Stöd för partitionering

Event Hubs har ett liknande koncept för fysisk partition som Kafka. Men till skillnad från Kafkas automatiska ombalansering mellan konsumenter och partitioner tillhandahåller Event Hubs ett slags förebyggande läge. Lagringskontot fungerar som en leasingmekanism för att fastställa vilken konsument som äger vilken partition. När en ny konsument startar försöker den stjäla vissa partitioner från de mest belastade konsumenterna för att uppnå arbetsbelastningsbalansen.

För att ange belastningsutjämningsstrategin tillhandahålls egenskaper för spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.load-balancing.*. Mer information finns i avsnittet Konsumentegenskaper.

Stöd för batchkonsumenter

Spring Cloud Azure Stream Event Hubs binder stöder Spring Cloud Stream Batch Consumer-funktion.

Om du vill arbeta med batch-konsumentläge ställer du in egenskapen spring.cloud.stream.bindings.<binding-name>.consumer.batch-mode på true. När det är aktiverat tas ett meddelande med en nyttolast av en lista över batchade händelser emot och skickas till funktionen Consumer. Varje meddelanderubrik konverteras också till en lista, där innehållet är det associerade rubrikvärdet som parsas från varje händelse. De gemensamma huvudena för partitions-ID, checkpointer och de senaste egenskaperna visas som ett enda värde eftersom hela batchen med händelser delar samma värde. Mer information finns i meddelanderubrikerna Event Hubs i Spring Cloud Azure-stöd för Spring Integration.

Obs!

Kontrollpunktshuvudet finns bara när MANUAL kontrollpunktsläge används.

Kontrollpunkter för batchkonsumenten stöder två lägen: BATCH och MANUAL. BATCH läget är ett automatiskt kontrollpunktsläge för att kontrollera hela batchen med händelser tillsammans när bindemedlet tar emot dem. MANUAL läge är att kontrollera händelserna av användare. När den används skickas Checkpointer till meddelandehuvudet och användarna kan använda den för att göra kontrollpunkter.

Du kan ange batchstorleken genom att ange egenskaperna max-size och max-wait-time som har prefixet spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.batch.. Egenskapen max-size är nödvändig och egenskapen max-wait-time är valfri. Mer information finns i avsnittet Konsumentegenskaper.

Konfiguration av beroenden

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-stream-binder-eventhubs</artifactId>
</dependency>

Du kan också använda Spring Cloud Azure Stream Event Hubs Starter, som du ser i följande exempel för Maven:

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

Konfiguration

Bindern innehåller följande tre delar av konfigurationsalternativ:

Egenskaper för anslutningskonfiguration

Det här avsnittet innehåller de konfigurationsalternativ som används för att ansluta till Azure Event Hubs.

Obs!

Om du väljer att använda ett säkerhetsobjekt för att autentisera och auktorisera med Microsoft Entra-ID för åtkomst till en Azure-resurs kan du läsa Auktorisera åtkomst med Microsoft Entra-ID för att se till att säkerhetsobjektet har beviljats tillräcklig behörighet för åtkomst till Azure-resursen.

Anslutningskonfigurerbara egenskaper för spring-cloud-azure-stream-binder-eventhubs:

Egenskap Typ Beskrivning
spring.cloud.azure.eventhubs.enabled boolesk Om en Azure Event Hubs är aktiverad.
spring.cloud.azure.eventhubs.connection-string Sträng Anslutningssträngsvärde för Event Hubs-namnområde.
spring.cloud.azure.eventhubs.namespace Sträng Event Hubs-namnområdesvärde, som är prefixet för det fullständiga domännamnet. Ett FQDN ska bestå av NamespaceName.DomainName
spring.cloud.azure.eventhubs.domain-name Sträng Domännamn för ett Namnområdesvärde för Azure Event Hubs.
spring.cloud.azure.eventhubs.custom-endpoint-address Sträng Anpassad slutpunktsadress.

Tips

Vanliga konfigurationsalternativ för Azure Service SDK kan konfigureras även för Spring Cloud Azure Stream Event Hubs-pärmen. Konfigurationsalternativen som stöds introduceras i Spring Cloud Azure-konfigurationenoch kan konfigureras med antingen det enhetliga prefixet spring.cloud.azure. eller prefixet för spring.cloud.azure.eventhubs..

Bindaren stöder också Spring Could Azure Resource Manager som standard. Information om hur du hämtar anslutningssträngen med säkerhetsobjekt som inte har beviljats de roller som är relaterade till Data finns i avsnittet Grundläggande användning i Spring Could Azure Resource Manager.

Konfigurationsegenskaper för kontrollpunkt

Det här avsnittet innehåller konfigurationsalternativen för Storage Blobs-tjänsten, som används för att bevara partitionsägarskap och kontrollpunktsinformation.

Obs!

Från version 4.0.0 gäller att om egenskapen för spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists inte aktiveras manuellt, skapas ingen lagringscontainer automatiskt med namnet från spring.cloud.stream.bindings.binding-name.destination.

Konfigurerbara egenskaper för kontrollpunkter för spring-cloud-azure-stream-binder-eventhubs:

Egenskap Typ Beskrivning
spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists boolesk Om skapande av containrar ska tillåtas om de inte redan finns.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name Sträng Namn på lagringskontot.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key Sträng Åtkomstnyckel för lagringskonto.
spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name Sträng Namn på lagringscontainer.

Tips

Vanliga konfigurationsalternativ för Azure Service SDK är även tillgängliga för konfiguration av Storage Blob-kontrollpunktsarkiv. Konfigurationsalternativen som stöds introduceras i Spring Cloud Azure-konfigurationenoch kan konfigureras med antingen det enhetliga prefixet spring.cloud.azure. eller prefixet för spring.cloud.azure.eventhubs.processor.checkpoint-store.

Konfigurationsegenskaper för Azure Event Hubs-bindning

Följande alternativ är indelade i fyra avsnitt: Konsumentegenskaper, Avancerade konsumentkonfigurationer, Producentegenskaper och Avancerade producentkonfigurationer.

Konsumentegenskaper

Dessa egenskaper exponeras via EventHubsConsumerProperties.

Obs!

För att undvika upprepning, eftersom version 4.17.0 och 5.11.0, stöder Spring Cloud Azure Stream Binder Event Hubs inställningsvärden för alla kanaler, i formatet spring.cloud.stream.eventhubs.default.consumer.<property>=<value>.

Konsumentkonfigurerbara egenskaper för spring-cloud-azure-stream-binder-eventhubs:

Egenskap Typ Beskrivning
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.checkpoint.mode CheckpointMode Kontrollpunktsläge som används när konsumenten bestämmer hur ett kontrollpunktsmeddelande ska visas
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.checkpoint.count Heltal Bestämmer hur många meddelanden varje partition ska ha för att utföra en kontrollpunkt. Börjar gälla endast när PARTITION_COUNT kontrollpunktsläge används.
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.checkpoint.interval Varaktighet Bestämmer tidsintervallet för att göra en kontrollpunkt. Börjar gälla endast när TIME kontrollpunktsläge används.
spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.batch.max-size Heltal Det maximala antalet händelser i en batch. Krävs för batch-konsumentläge.
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.batch.max-wait-time Varaktighet Den maximala tidsperioden för batchförbrukning. Börjar gälla endast när batch-konsumentläget är aktiverat och är valfritt.
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.load-balancing.update-interval Varaktighet Tidsintervall för uppdatering.
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.load-balancing.strategy Belastningsutjämningsstrategi Belastningsutjämningsstrategin.
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.load-balancing.partition-ownership-expiration-interval Varaktighet Tidsåtgången efter vilken ägarskapet för partitionen upphör att gälla.
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.track-last-enqueued-event-properties boolesk Om händelseprocessorn ska begära information om den senaste begärda händelsen på den associerade partitionen och spåra den informationen när händelser tas emot.
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.prefetch-count Heltal Det antal som konsumenten använder för att styra hur många händelser som Event Hub-konsumenten aktivt tar emot och lägger i lokal kö.
spring.cloud.stream.eventhubs.bindings.binding-name.consumer.initial-partition-event-position Mappa med nyckeln som partitions-ID och värden för StartPositionProperties Mappningen som innehåller den händelsepositionen som ska användas för varje partition om det inte finns någon kontrollpunkt för partitionen i kontrollpunktslagret. Den här mappningen baseras på partitions-ID:t.

Obs!

Konfigurationen initial-partition-event-position accepterar en map för att ange den första positionen för varje händelsehubb. Därför är dess nyckel partitions-ID och värdet är StartPositionProperties, som innehåller egenskaper för offset, sekvensnummer, köad datumtid och huruvida inkluderande. Du kan till exempel ange den som

spring:
  cloud:
    stream:
      eventhubs:
        bindings:
          <binding-name>:
            consumer:
              initial-partition-event-position:
                0:
                  offset: earliest
                1:
                  sequence-number: 100
                2:
                  enqueued-date-time: 2022-01-12T13:32:47.650005Z
                4:
                  inclusive: false
Avancerad konsumentkonfiguration

Ovanstående konfigurationer för anslutning, kontrollpunkt och gemensam Azure SDKs-klient stöder anpassning för varje bindningskonsument, vilket du kan konfigurera med prefixet spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer..

Producentegenskaper

Dessa egenskaper exponeras via EventHubsProducerProperties.

Obs!

För att undvika upprepning, eftersom version 4.17.0 och 5.11.0, stöder Spring Cloud Azure Stream Binder Event Hubs inställningsvärden för alla kanaler, i formatet spring.cloud.stream.eventhubs.default.producer.<property>=<value>.

Producentkonfigurerbara egenskaper för spring-cloud-azure-stream-binder-eventhubs:

Egenskap Typ Beskrivning
spring.cloud.stream.eventhubs.bindings.binding-name.producer.sync boolesk Switch-flaggan för synkronisering av producent. Om detta är sant kommer producenten att vänta på ett svar efter en sändningsoperation.
spring.cloud.stream.eventhubs.bindings.binding-name.producer.send-timeout lång Hur lång tid det går att vänta på ett svar efter en sändningsåtgärd. Börjar gälla endast när en synkroniseringsproducent är aktiverad.
Avancerad konfiguration för producent

Ovanstående anslutning och vanliga Azure SDKs-klient konfiguration stöder anpassning för varje pärmproducent, som du kan konfigurera med prefixet spring.cloud.stream.eventhubs.bindings.<binding-name>.producer..

Grundläggande användning

Skicka och ta emot meddelanden från/till Event Hubs

  1. Fyll i konfigurationsalternativen med information om autentiseringsuppgifter.

    • För autentiseringsuppgifter som anslutningssträng konfigurerar du följande egenskaper i filen application.yml:

      spring:
  cloud:
    azure:
      eventhubs:
        connection-string: ${EVENTHUB_NAMESPACE_CONNECTION_STRING}
        processor:
          checkpoint-store:
            container-name: ${CHECKPOINT_CONTAINER}
            account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
            account-key: ${CHECKPOINT_ACCESS_KEY}
    function:
      definition: consume;supply
    stream:
      bindings:
        consume-in-0:
          destination: ${EVENTHUB_NAME}
          group: ${CONSUMER_GROUP}
        supply-out-0:
          destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
      eventhubs:
        bindings:
          consume-in-0:
            consumer:
              checkpoint:
                mode: MANUAL

      Obs!

      Microsoft rekommenderar att du använder det säkraste tillgängliga autentiseringsflödet. Det autentiseringsflöde som beskrivs i den här proceduren, till exempel för databaser, cacheminnen, meddelanden eller AI-tjänster, kräver en mycket hög grad av förtroende för programmet och medför risker som inte finns i andra flöden. Använd endast det här flödet när säkrare alternativ, till exempel hanterade identiteter för lösenordslösa eller nyckellösa anslutningar, inte är genomförbara. För lokala datoråtgärder föredrar du användaridentiteter för lösenordslösa eller nyckellösa anslutningar.

    • För autentiseringsuppgifter för en service principal konfigurerar du följande egenskaper i din application.yml-fil:

      spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: <tenant>
      eventhubs:
        namespace: ${EVENTHUB_NAMESPACE}
        processor:
          checkpoint-store:
            container-name: ${CONTAINER_NAME}
            account-name: ${ACCOUNT_NAME}
    function:
      definition: consume;supply
    stream:
      bindings:
        consume-in-0:
          destination: ${EVENTHUB_NAME}
          group: ${CONSUMER_GROUP}
        supply-out-0:
          destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
      eventhubs:
        bindings:
          consume-in-0:
            consumer:
              checkpoint:
                mode: MANUAL

Obs!

De värden som tillåts för tenant-id är: common, organizations, consumerseller klientorganisations-ID. Mer information om de här värdena finns i avsnittet Fel slutpunkt användes (personliga konton och organisationskonton) i Fel AADSTS50020 – Användarkonto från identitetsleverantören finns inte i klientorganisationen. Information om hur du konverterar din app för en klientorganisation finns i Konvertera enklientapp till flera klientorganisationer på Microsoft Entra-ID.

  • För autentiseringsuppgifter som hanterade identiteter konfigurerar du följande egenskaper i din application.yml-fil:

    spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${AZURE_MANAGED_IDENTITY_CLIENT_ID} # Only needed when using a user-assigned managed identity
      eventhubs:
        namespace: ${EVENTHUB_NAMESPACE}
        processor:
          checkpoint-store:
            container-name: ${CONTAINER_NAME}
            account-name: ${ACCOUNT_NAME}
    function:
      definition: consume;supply
    stream:
      bindings:
        consume-in-0:
          destination: ${EVENTHUB_NAME}
          group: ${CONSUMER_GROUP}
        supply-out-0:
          destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}

      eventhubs:
        bindings:
          consume-in-0:
            consumer:
              checkpoint:
                mode: MANUAL

  1. Definiera leverantör och konsument.

    @Bean
public Consumer<Message<String>> consume() {
    return message -> {
        Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
        LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                message.getPayload(),
                message.getHeaders().get(EventHubsHeaders.PARTITION_KEY),
                message.getHeaders().get(EventHubsHeaders.SEQUENCE_NUMBER),
                message.getHeaders().get(EventHubsHeaders.OFFSET),
                message.getHeaders().get(EventHubsHeaders.ENQUEUED_TIME)
        );

        checkpointer.success()
                .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                .doOnError(error -> LOGGER.error("Exception found", error))
                .block();
    };
}

@Bean
public Supplier<Message<String>> supply() {
    return () -> {
        LOGGER.info("Sending message, sequence " + i);
        return MessageBuilder.withPayload("Hello world, " + i++).build();
    };
}

Stöd för partitionering

En PartitionSupplier med partitionsinformation från användaren skapas för att konfigurera partitionsinformationen om meddelandet som ska skickas. Följande flödesschema visar processen för att hämta olika prioriteringar för partitions-ID och nyckel:

diagram som visar ett flödesschema för partitioneringsstödprocessen.

Stöd för batchkonsumenter

  1. Ange konfigurationsalternativen för batchen enligt följande exempel:

    spring:
  cloud:
    function:
      definition: consume
    stream:
      bindings:
        consume-in-0:
          destination: ${AZURE_EVENTHUB_NAME}
          group: ${AZURE_EVENTHUB_CONSUMER_GROUP}
          consumer:
            batch-mode: true
      eventhubs:
        bindings:
          consume-in-0:
            consumer:
              batch:
                max-batch-size: 10 # Required for batch-consumer mode
                max-wait-time: 1m # Optional, the default value is null
              checkpoint:
                mode: BATCH # or MANUAL as needed

  2. Definiera leverantör och konsument.

    För kontrollpunktsläge som BATCHkan du använda följande kod för att skicka meddelanden och använda i batchar.

    @Bean
public Consumer<Message<List<String>>> consume() {
    return message -> {
        for (int i = 0; i < message.getPayload().size(); i++) {
            LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                    message.getPayload().get(i),
                    ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_PARTITION_KEY)).get(i),
                    ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_SEQUENCE_NUMBER)).get(i),
                    ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_OFFSET)).get(i),
                    ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_ENQUEUED_TIME)).get(i));
        }
    };
}

@Bean
public Supplier<Message<String>> supply() {
    return () -> {
        LOGGER.info("Sending message, sequence " + i);
        return MessageBuilder.withPayload("\"test"+ i++ +"\"").build();
    };
}

    För checkpointläge inställt på MANUAL kan du använda följande kod för att skicka meddelanden och ta emot samt checkpointa batchvis.

    @Bean
public Consumer<Message<List<String>>> consume() {
    return message -> {
        for (int i = 0; i < message.getPayload().size(); i++) {
            LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                message.getPayload().get(i),
                ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_PARTITION_KEY)).get(i),
                ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_SEQUENCE_NUMBER)).get(i),
                ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_OFFSET)).get(i),
                ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_ENQUEUED_TIME)).get(i));
        }

        Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
        checkpointer.success()
                    .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                    .doOnError(error -> LOGGER.error("Exception found", error))
                    .block();
    };
}

@Bean
public Supplier<Message<String>> supply() {
    return () -> {
        LOGGER.info("Sending message, sequence " + i);
        return MessageBuilder.withPayload("\"test"+ i++ +"\"").build();
    };
}

Obs!

I batchförbrukningsläget är standardinnehållstypen för Spring Cloud Stream-pärm application/json, så se till att meddelandets nyttolast är anpassad till innehållstypen. När du till exempel använder standardinnehållstypen för application/json för att ta emot meddelanden med String nyttolast ska nyttolasten vara JSON String, omgiven av dubbla citattecken för den ursprungliga String texten. För text/plain innehållstyp kan det vara ett String objekt direkt. Mer information finns i Spring Cloud Stream Content Type Negotiation.

Hantera felmeddelanden

  • Hantera utgående bindningsfelmeddelanden

    Som standard skapar Spring Integration en global felkanal med namnet errorChannel. Konfigurera följande meddelandeslutpunkt för att hantera utgående bindningsfelmeddelanden.

    @ServiceActivator(inputChannel = IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)
public void handleError(ErrorMessage message) {
    LOGGER.error("Handling outbound binding error: " + message);
}

  • Hantera inkommande bindningsfelmeddelanden

    Spring Cloud Stream Event Hubs Binder stöder en lösning för att hantera fel för inkommande meddelandebindningar: felhanterare.

    Felhanterare:

    Spring Cloud Stream exponerar en mekanism som du kan använda för att tillhandahålla en anpassad felhanterare genom att lägga till en Consumer som accepterar ErrorMessage instanser. Mer information finns i Hantera felmeddelanden i Spring Cloud Stream-dokumentationen.

    • standardfelhanterare för bindning

      Konfigurera en enskild Consumer bean för att ta emot alla inkommande bindningsfelmeddelanden. Följande standardfunktion prenumererar på varje inkommande bindningsfelkanal:

      @Bean
public Consumer<ErrorMessage> myDefaultHandler() {
    return message -> {
        // consume the error message
    };
}

      Du måste också ange egenskapen spring.cloud.stream.default.error-handler-definition till funktionsnamnet.

    • Bindningsspecifik felhanterare

      Konfigurera en Consumer bean för att ta emot de specifika inkommande bindningsfelmeddelandena. Följande funktion prenumererar på den specifika felkanalen för den inkommande bindningen och har högre prioritet än bindningens standardfelhanterare:

      @Bean
public Consumer<ErrorMessage> myErrorHandler() {
    return message -> {
        // consume the error message
    };
}

      Du måste också ange egenskapen spring.cloud.stream.bindings.<input-binding-name>.error-handler-definition till funktionsnamnet.

Meddelandehuvuden för Event Hubs

De grundläggande meddelandehuvuden som stöds finns i meddelanderubrikerna Event Hubs i Spring Cloud Azure-stöd för Spring Integration.

Stöd för flera bindningar

Anslutning till flera Event Hubs-namnområden stöds också med hjälp av flera pärmar. Det här exemplet tar en anslutningssträng som exempel. Autentiseringsuppgifter för tjänsthuvudnamn och hanterade identiteter stöds också. Du kan ange relaterade egenskaper i varje bindemedels miljöinställningar.

  1. Om du vill använda flera pärmar med Event Hubs konfigurerar du följande egenskaper i din application.yml-fil:

    spring:
  cloud:
    function:
      definition: consume1;supply1;consume2;supply2
    stream:
      bindings:
        consume1-in-0:
          destination: ${EVENTHUB_NAME_01}
          group: ${CONSUMER_GROUP_01}
        supply1-out-0:
          destination: ${THE_SAME_EVENTHUB_NAME_01_AS_ABOVE}
        consume2-in-0:
          binder: eventhub-2
          destination: ${EVENTHUB_NAME_02}
          group: ${CONSUMER_GROUP_02}
        supply2-out-0:
          binder: eventhub-2
          destination: ${THE_SAME_EVENTHUB_NAME_02_AS_ABOVE}
      binders:
        eventhub-1:
          type: eventhubs
          default-candidate: true
          environment:
            spring:
              cloud:
                azure:
                  eventhubs:
                    connection-string: ${EVENTHUB_NAMESPACE_01_CONNECTION_STRING}
                    processor:
                      checkpoint-store:
                        container-name: ${CHECKPOINT_CONTAINER_01}
                        account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                        account-key: ${CHECKPOINT_ACCESS_KEY}
        eventhub-2:
          type: eventhubs
          default-candidate: false
          environment:
            spring:
              cloud:
                azure:
                  eventhubs:
                    connection-string: ${EVENTHUB_NAMESPACE_02_CONNECTION_STRING}
                    processor:
                      checkpoint-store:
                        container-name: ${CHECKPOINT_CONTAINER_02}
                        account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                        account-key: ${CHECKPOINT_ACCESS_KEY}
      eventhubs:
        bindings:
          consume1-in-0:
            consumer:
              checkpoint:
                mode: MANUAL
          consume2-in-0:
            consumer:
              checkpoint:
                mode: MANUAL
      poller:
        initial-delay: 0
        fixed-delay: 1000

    Obs!

    Den föregående applikationsfilen visar hur du konfigurerar en enda standardpoller som ska tillämpas på alla bindningar. Om du vill konfigurera polleraren för en specifik bindning kan du använda en konfiguration som spring.cloud.stream.bindings.<binding-name>.producer.poller.fixed-delay=3000.

    Obs!

    Microsoft rekommenderar att du använder det säkraste tillgängliga autentiseringsflödet. Det autentiseringsflöde som beskrivs i den här proceduren, till exempel för databaser, cacheminnen, meddelanden eller AI-tjänster, kräver en mycket hög grad av förtroende för programmet och medför risker som inte finns i andra flöden. Använd endast det här flödet när säkrare alternativ, till exempel hanterade identiteter för lösenordslösa eller nyckellösa anslutningar, inte är genomförbara. För lokala datoråtgärder föredrar du användaridentiteter för lösenordslösa eller nyckellösa anslutningar.

  2. Vi behöver definiera två leverantörer och två konsumenter:

    @Bean
public Supplier<Message<String>> supply1() {
    return () -> {
        LOGGER.info("Sending message1, sequence1 " + i);
        return MessageBuilder.withPayload("Hello world1, " + i++).build();
    };
}

@Bean
public Supplier<Message<String>> supply2() {
    return () -> {
        LOGGER.info("Sending message2, sequence2 " + j);
        return MessageBuilder.withPayload("Hello world2, " + j++).build();
    };
}

@Bean
public Consumer<Message<String>> consume1() {
    return message -> {
        Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
        LOGGER.info("New message1 received: '{}'", message);
        checkpointer.success()
                .doOnSuccess(success -> LOGGER.info("Message1 '{}' successfully checkpointed", message))
                .doOnError(error -> LOGGER.error("Exception found", error))
                .block();
    };
}

@Bean
public Consumer<Message<String>> consume2() {
    return message -> {
        Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
        LOGGER.info("New message2 received: '{}'", message);
        checkpointer.success()
                .doOnSuccess(success -> LOGGER.info("Message2 '{}' successfully checkpointed", message))
                .doOnError(error -> LOGGER.error("Exception found", error))
                .block();
    };
}

Resurstilldelning

Event Hubs binder stöder etablering av händelsehubb och konsumentgrupp. Användarna kan använda följande egenskaper för att aktivera etablering.

spring:
  cloud:
    azure:
      credential:
        tenant-id: <tenant>
      profile:
        subscription-id: ${AZURE_SUBSCRIPTION_ID}
      eventhubs:
        resource:
          resource-group: ${AZURE_EVENTHUBS_RESOURCE_GROUP}

Obs!

De värden som tillåts för tenant-id är: common, organizations, consumerseller klientorganisations-ID. Mer information om de här värdena finns i avsnittet Fel slutpunkt användes (personliga konton och organisationskonton) i Fel AADSTS50020 – Användarkonto från identitetsleverantören finns inte i klientorganisationen. Information om hur du konverterar din app för en klientorganisation finns i Konvertera enklientapp till flera klientorganisationer på Microsoft Entra-ID.

Exempel

Mer information finns i lagringsplatsen azure-spring-boot-samples på GitHub.

Spring Cloud Stream Binder för Azure Service Bus

Viktiga begrepp

Spring Cloud Stream Binder för Azure Service Bus tillhandahåller bindningsimplementeringen för Spring Cloud Stream Framework. Den här implementeringen använder Spring Integration Service Bus-kanaladaptrar som grund.

Schemalagt meddelande

Den här pärmen stöder sändning av meddelanden till ett ämne för fördröjd bearbetning. Användare kan skicka schemalagda meddelanden med rubrik x-delay uttrycker i millisekunder en fördröjningstid för meddelandet. Meddelandet levereras till respektive avsnitt efter x-delay millisekunder.

Konsumentgrupp

Service Bus-ämnet ger liknande stöd för konsumentgruppen som Apache Kafka, men med lite annan logik. Den här bindaren förlitar sig på Subscription i ett ämne för att agera som en konsumentgrupp.

Konfiguration av beroenden

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-stream-binder-servicebus</artifactId>
</dependency>

Du kan också använda Spring Cloud Azure Stream Service Bus Starter, som du ser i följande exempel för Maven:

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

Konfiguration

Bindaren erbjuder följande två delar med konfigurationsalternativ:

Egenskaper för anslutningskonfiguration

Det här avsnittet innehåller de konfigurationsalternativ som används för att ansluta till Azure Service Bus.

Obs!

Om du väljer att använda ett säkerhetsobjekt för att autentisera och auktorisera med Microsoft Entra-ID för åtkomst till en Azure-resurs kan du läsa Auktorisera åtkomst med Microsoft Entra-ID för att se till att säkerhetsobjektet har beviljats tillräcklig behörighet för åtkomst till Azure-resursen.

Anslutningskonfigurerbara egenskaper för spring-cloud-azure-stream-binder-servicebus:

Egenskap Typ Beskrivning
spring.cloud.azure.servicebus.enabled boolesk Om en Azure Service Bus är aktiverad.
spring.cloud.azure.servicebus.connection-string Sträng Anslutningssträngsvärde för Service Bus-namnområde.
spring.cloud.azure.servicebus.custom-endpoint-address Sträng Den anpassade slutpunktsadress som ska användas vid anslutning till Service Bus.
spring.cloud.azure.servicebus.namespace Sträng Service Bus-namnområdesvärde, som är prefixet för FQDN. Ett FQDN ska bestå av NamespaceName.DomainName
spring.cloud.azure.servicebus.domain-name Sträng Domännamn för ett Azure Service Bus-namnområdesvärde.

Obs!

Vanliga konfigurationsalternativ för Azure Service SDK kan konfigureras även för Spring Cloud Azure Stream Service Bus-pärmen. Konfigurationsalternativen som stöds introduceras i Spring Cloud Azure-konfigurationenoch kan konfigureras med antingen det enhetliga prefixet spring.cloud.azure. eller prefixet för spring.cloud.azure.servicebus..

Bindaren stöder också Spring Could Azure Resource Manager som standard. Information om hur du hämtar anslutningssträngen med säkerhetsobjekt som inte har beviljats de roller som är relaterade till Data finns i avsnittet Grundläggande användning i Spring Could Azure Resource Manager.

Konfigurationsegenskaper för Azure Service Bus-bindning

Följande alternativ är indelade i fyra avsnitt: Konsumentegenskaper, Avancerade konsumentkonfigurationer, Producentegenskaper och Avancerade producentkonfigurationer.

Konsumentegenskaper

Dessa egenskaper exponeras via ServiceBusConsumerProperties.

Obs!

För att undvika upprepning, eftersom version 4.17.0 och 5.11.0, stöder Spring Cloud Azure Stream Binder Service Bus inställningsvärden för alla kanaler, i formatet spring.cloud.stream.servicebus.default.consumer.<property>=<value>.

Konsumentkonfigurerbara egenskaper för spring-cloud-azure-stream-binder-servicebus:

Egenskap Typ Standard Beskrivning
spring.cloud.stream.servicebus.bindings.binding-name.consumer.requeue-rejected boolesk falsk Om de misslyckade meddelandena dirigeras till DLQ.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.max-concurrent-calls Heltal 1 Maximalt antal samtidiga meddelanden som Service Bus-processorklienten ska bearbeta. När sessionen är aktiverad gäller den för varje session.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.max-concurrent-sessions Heltal noll Maximalt antal samtidiga sessioner som ska bearbetas vid en viss tidpunkt.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.session-enabled boolesk noll Om sessionen är aktiverad.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.session-idle-timeout Varaktighet noll Anger den maximala tiden (varaktighet) för att vänta tills ett meddelande tas emot för den aktiva sessionen.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.prefetch-count Heltal 0 Antalet prefetch för Service Bus-processorklienten.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.sub-queue Underkö ingen Typen av underkö som ska anslutas till.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.max-auto-lock-renew-duration Varaktighet 5 m Hur lång tid det går att fortsätta förnya låset automatiskt.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.receive-mode ServiceBusReceiveMode peek_lock Mottagningsläget för Service Bus-processorklienten.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.auto-complete boolesk sann Om meddelanden ska regleras automatiskt. Om det anges som falskt läggs ett meddelandehuvud för Checkpointer till för att göra det möjligt för utvecklare att reglera meddelanden manuellt.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.max-size-in-megabytes Lång 1024 Den maximala storleken på kön/ämnet i megabyte, vilket är storleken på det minne som allokerats för kön/ämnet.
spring.cloud.stream.servicebus.bindings.binding-name.consumer.default-message-time-to-live Varaktighet P10675199DT2H48M5.4775807S. (10675199 dagar, 2 timmar, 48 minuter, 5 sekunder och 477 millisekunder) Varaktigheten efter vilken meddelandet upphör att gälla, från och med när meddelandet skickas till Service Bus.

Viktig

När du använder Azure Resource Manager (ARM) måste du konfigurera egenskapen spring.cloud.stream.servicebus.bindings.<binding-name>.consume.entity-type. Mer information finns i servicebus-queue-binder-arm exemplet på GitHub.

Avancerad konsumentkonfiguration

Konfigurationerna ovan för anslutning och den gemensamma Azure SDKs-klienten stöder anpassning för varje binder-konsument, som kan konfigureras med prefixet spring.cloud.stream.servicebus.bindings.<binding-name>.consumer..

Producentegenskaper

Dessa egenskaper exponeras via ServiceBusProducerProperties.

Obs!

För att undvika upprepning, eftersom version 4.17.0 och 5.11.0, stöder Spring Cloud Azure Stream Binder Service Bus inställningsvärden för alla kanaler, i formatet spring.cloud.stream.servicebus.default.producer.<property>=<value>.

Producentkonfigurerbara egenskaper för spring-cloud-azure-stream-binder-servicebus:

Egenskap Typ Standard Beskrivning
spring.cloud.stream.servicebus.bindings.binding-name.producer.sync boolesk falsk Växla flagga för synkronisering av producent.
spring.cloud.stream.servicebus.bindings.binding-name.producer.send-timeout lång 10 000 Timeoutvärde för producentens sändning.
spring.cloud.stream.servicebus.bindings.binding-name.producer.entity-type ServiceBusEntityType noll Service Bus-enhetstyp för producenten, krävs för bindningsproducenten.
spring.cloud.stream.servicebus.bindings.binding-name.producer.max-size-in-megabytes Lång 1024 Den maximala storleken på kön/ämnet i megabyte, vilket är storleken på det minne som allokerats för kön/ämnet.
spring.cloud.stream.servicebus.bindings.binding-name.producer.default-message-time-to-live Varaktighet P10675199DT2H48M5.4775807S. (10675199 dagar, 2 timmar, 48 minuter, 5 sekunder och 477 millisekunder) Varaktigheten efter vilken meddelandet upphör att gälla, från och med när meddelandet skickas till Service Bus.

Viktig

När du använder bindningsproducenten måste egenskapen för spring.cloud.stream.servicebus.bindings.<binding-name>.producer.entity-type konfigureras.

Avancerad konfiguration för producent

Ovanstående anslutning och vanliga Azure SDKs-klient konfiguration stöder anpassning för varje pärmproducent, som du kan konfigurera med prefixet spring.cloud.stream.servicebus.bindings.<binding-name>.producer..

Grundläggande användning

Skicka och ta emot meddelanden från/till Service Bus

  1. Fyll i konfigurationsalternativen med information om autentiseringsuppgifter.

    • För autentiseringsuppgifter som anslutningssträng konfigurerar du följande egenskaper i filen application.yml:

      spring:
   cloud:
     azure:
       servicebus:
         connection-string: ${SERVICEBUS_NAMESPACE_CONNECTION_STRING}
     function:
       definition: consume;supply
     stream:
       bindings:
         consume-in-0:
           destination: ${SERVICEBUS_ENTITY_NAME}
           # If you use Service Bus Topic, add the following configuration
           # group: ${SUBSCRIPTION_NAME}
         supply-out-0:
           destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
       servicebus:
         bindings:
           consume-in-0:
             consumer:
               auto-complete: false
           supply-out-0:
             producer:
               entity-type: queue # set as "topic" if you use Service Bus Topic

      Obs!

      Microsoft rekommenderar att du använder det säkraste tillgängliga autentiseringsflödet. Det autentiseringsflöde som beskrivs i den här proceduren, till exempel för databaser, cacheminnen, meddelanden eller AI-tjänster, kräver en mycket hög grad av förtroende för programmet och medför risker som inte finns i andra flöden. Använd endast det här flödet när säkrare alternativ, till exempel hanterade identiteter för lösenordslösa eller nyckellösa anslutningar, inte är genomförbara. För lokala datoråtgärder föredrar du användaridentiteter för lösenordslösa eller nyckellösa anslutningar.

    • För autentiseringsuppgifter för en service principal konfigurerar du följande egenskaper i din application.yml-fil:

      spring:
   cloud:
     azure:
       credential:
         client-id: ${AZURE_CLIENT_ID}
         client-secret: ${AZURE_CLIENT_SECRET}
       profile:
         tenant-id: <tenant>
       servicebus:
         namespace: ${SERVICEBUS_NAMESPACE}
     function:
       definition: consume;supply
     stream:
       bindings:
         consume-in-0:
           destination: ${SERVICEBUS_ENTITY_NAME}
           # If you use Service Bus Topic, add the following configuration
           # group: ${SUBSCRIPTION_NAME}
         supply-out-0:
           destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
       servicebus:
         bindings:
           consume-in-0:
             consumer:
               auto-complete: false
           supply-out-0:
             producer:
               entity-type: queue # set as "topic" if you use Service Bus Topic

Obs!

De värden som tillåts för tenant-id är: common, organizations, consumerseller klientorganisations-ID. Mer information om de här värdena finns i avsnittet Fel slutpunkt användes (personliga konton och organisationskonton) i Fel AADSTS50020 – Användarkonto från identitetsleverantören finns inte i klientorganisationen. Information om hur du konverterar din app för en klientorganisation finns i Konvertera enklientapp till flera klientorganisationer på Microsoft Entra-ID.

  • För autentiseringsuppgifter som hanterade identiteter konfigurerar du följande egenskaper i din application.yml-fil:

    spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${MANAGED_IDENTITY_CLIENT_ID} # Only needed when using a user-assigned managed identity
      servicebus:
        namespace: ${SERVICEBUS_NAMESPACE}
    function:
      definition: consume;supply
    stream:
      bindings:
        consume-in-0:
          destination: ${SERVICEBUS_ENTITY_NAME}
          # If you use Service Bus Topic, add the following configuration
          # group: ${SUBSCRIPTION_NAME}
        supply-out-0:
          destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
      servicebus:
        bindings:
          consume-in-0:
            consumer:
              auto-complete: false
          supply-out-0:
            producer:
              entity-type: queue # set as "topic" if you use Service Bus Topic

  1. Definiera leverantör och konsument.

    @Bean
public Consumer<Message<String>> consume() {
    return message -> {
        Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
        LOGGER.info("New message received: '{}'", message.getPayload());

        checkpointer.success()
                .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                .doOnError(error -> LOGGER.error("Exception found", error))
                .block();
    };
}

@Bean
public Supplier<Message<String>> supply() {
    return () -> {
        LOGGER.info("Sending message, sequence " + i);
        return MessageBuilder.withPayload("Hello world, " + i++).build();
    };
}

Stöd för partitionsnyckel

Pärmen stöder Service Bus-partitionering genom att tillåta inställning av partitionsnyckel och sessions-ID i meddelandehuvudet. I det här avsnittet beskrivs hur du anger partitionsnyckel för meddelanden.

Spring Cloud Stream tillhandahåller en spEL-uttrycksegenskap för partitionsnyckeln spring.cloud.stream.bindings.<binding-name>.producer.partition-key-expression. Du kan till exempel ange den här egenskapen som "'partitionKey-' + headers[<message-header-key>]" och lägga till ett huvud med namnet message-header-key. Spring Cloud Stream använder värdet för denna header när uttrycket utvärderas för att fastställa en partitionsnyckel. Följande kod ger en exempelproducent:

@Bean
public Supplier<Message<String>> generate() {
    return () -> {
        String value = "random payload";
        return MessageBuilder.withPayload(value)
            .setHeader("<message-header-key>", value.length() % 4)
            .build();
    };
}

Sessionsstöd

Bindningen stöder meddelandesessioner för Service Bus. Sessions-ID för ett meddelande kan anges via meddelanderubriken.

@Bean
public Supplier<Message<String>> generate() {
    return () -> {
        String value = "random payload";
        return MessageBuilder.withPayload(value)
            .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session ID")
            .build();
    };
}

Obs!

Enligt Service Bus-partitioneringhar sessions-ID högre prioritet än partitionsnyckel. Så när både ServiceBusMessageHeaders#SESSION_ID och ServiceBusMessageHeaders#PARTITION_KEY huvuden anges används värdet för sessions-ID:t så småningom för att skriva över värdet för partitionsnyckeln.

Hantera felmeddelanden

  • Hantera utgående bindningsfelmeddelanden

    Som standard skapar Spring Integration en global felkanal med namnet errorChannel. Konfigurera följande meddelandeslutpunkt för att hantera utgående bindningsfel.

    @ServiceActivator(inputChannel = IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)
public void handleError(ErrorMessage message) {
    LOGGER.error("Handling outbound binding error: " + message);
}

  • Hantera inkommande bindningsfelmeddelanden

    Spring Cloud Stream Service Bus Binder stöder två lösningar för att hantera fel för inkommande meddelandebindningar: binder-felhanteraren och hanterare.

    Binder-felhanterare:

    Standardhanteraren för bindemedelsfel hanterar den inkommande bindningen. Du använder den här hanteraren för att skicka meddelanden som misslyckats till dead letter-kön när spring.cloud.stream.servicebus.bindings.<binding-name>.consumer.requeue-rejected är aktiverat. Annars avbryts de misslyckade meddelandena. Felfelsökaren för bindaren kan inte användas tillsammans med andra angivna felhanterare.

    Felhanterare:

    Spring Cloud Stream exponerar en mekanism som du kan använda för att tillhandahålla en anpassad felhanterare genom att lägga till en Consumer som accepterar ErrorMessage instanser. Mer information finns i Hantera felmeddelanden i Spring Cloud Stream-dokumentationen.

    • standardfelhanterare för bindning

      Konfigurera en enskild Consumer bean för att ta emot alla inkommande bindningsfelmeddelanden. Följande standardfunktion prenumererar på varje inkommande bindningsfelkanal:

      @Bean
public Consumer<ErrorMessage> myDefaultHandler() {
    return message -> {
        // consume the error message
    };
}

      Du måste också ange egenskapen spring.cloud.stream.default.error-handler-definition till funktionsnamnet.

    • Bindningsspecifik felhanterare

      Konfigurera en Consumer bean för att ta emot de specifika inkommande bindningsfelmeddelandena. Följande funktion prenumererar på felkanalen för den specifika inkommande bindningen med högre prioritet än standardfelhanteraren för bindningen.

      @Bean
public Consumer<ErrorMessage> myDefaultHandler() {
    return message -> {
        // consume the error message
    };
}

      Du måste också ange egenskapen spring.cloud.stream.bindings.<input-binding-name>.error-handler-definition till funktionsnamnet.

Service Bus-meddelandehuvuden

De grundläggande meddelandehuvuden som stöds finns i avsnittet Service Bus-meddelandehuvuden i Spring Cloud Azure-stöd för Spring Integration.

Obs!

När du ställer in partitionsnyckeln är prioriteten för meddelandehuvudet högre än spring cloud stream-egenskapen. Så spring.cloud.stream.bindings.<binding-name>.producer.partition-key-expression börjar gälla endast när ingen av ServiceBusMessageHeaders#SESSION_ID- och ServiceBusMessageHeaders#PARTITION_KEY-rubrikerna har konfigurerats.

Stöd för flera bindningar

Anslutning till flera Service Bus-namnområden stöds också med hjälp av flera pärmar. Det här exemplet tar anslutningssträngen som exempel. Autentiseringsuppgifter för tjänstens huvudnamn och hanterade identiteter stöds också, användare kan ange relaterade egenskaper i varje pärms miljöinställningar.

  1. Om du vill använda flera bindemedel för ServiceBus konfigurerar du följande egenskaper i din application.yml-fil:

    spring:
  cloud:
    function:
      definition: consume1;supply1;consume2;supply2
    stream:
      bindings:
        consume1-in-0:
          destination: ${SERVICEBUS_TOPIC_NAME}
          group: ${SUBSCRIPTION_NAME}
        supply1-out-0:
          destination: ${SERVICEBUS_TOPIC_NAME_SAME_AS_ABOVE}
        consume2-in-0:
          binder: servicebus-2
          destination: ${SERVICEBUS_QUEUE_NAME}
        supply2-out-0:
          binder: servicebus-2
          destination: ${SERVICEBUS_QUEUE_NAME_SAME_AS_ABOVE}
      binders:
        servicebus-1:
          type: servicebus
          default-candidate: true
          environment:
            spring:
              cloud:
                azure:
                  servicebus:
                    connection-string: ${SERVICEBUS_NAMESPACE_01_CONNECTION_STRING}
        servicebus-2:
          type: servicebus
          default-candidate: false
          environment:
            spring:
              cloud:
                azure:
                  servicebus:
                    connection-string: ${SERVICEBUS_NAMESPACE_02_CONNECTION_STRING}
      servicebus:
        bindings:
          consume1-in-0:
            consumer:
              auto-complete: false
          supply1-out-0:
            producer:
              entity-type: topic
          consume2-in-0:
            consumer:
              auto-complete: false
          supply2-out-0:
            producer:
              entity-type: queue
      poller:
        initial-delay: 0
        fixed-delay: 1000

    Obs!

    Den föregående applikationsfilen visar hur du konfigurerar en enda standardpoller som ska tillämpas på alla bindningar. Om du vill konfigurera polleraren för en specifik bindning kan du använda en konfiguration som spring.cloud.stream.bindings.<binding-name>.producer.poller.fixed-delay=3000.

    Obs!

    Microsoft rekommenderar att du använder det säkraste tillgängliga autentiseringsflödet. Det autentiseringsflöde som beskrivs i den här proceduren, till exempel för databaser, cacheminnen, meddelanden eller AI-tjänster, kräver en mycket hög grad av förtroende för programmet och medför risker som inte finns i andra flöden. Använd endast det här flödet när säkrare alternativ, till exempel hanterade identiteter för lösenordslösa eller nyckellösa anslutningar, inte är genomförbara. För lokala datoråtgärder föredrar du användaridentiteter för lösenordslösa eller nyckellösa anslutningar.

  2. vi behöver definiera två leverantörer och två konsumenter

    @Bean
public Supplier<Message<String>> supply1() {
    return () -> {
        LOGGER.info("Sending message1, sequence1 " + i);
        return MessageBuilder.withPayload("Hello world1, " + i++).build();
    };
}

@Bean
public Supplier<Message<String>> supply2() {
    return () -> {
        LOGGER.info("Sending message2, sequence2 " + j);
        return MessageBuilder.withPayload("Hello world2, " + j++).build();
    };
}

@Bean
public Consumer<Message<String>> consume1() {
    return message -> {
        Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
        LOGGER.info("New message1 received: '{}'", message);
        checkpointer.success()
                .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                .doOnError(e -> LOGGER.error("Error found", e))
                .block();
    };
}

@Bean
public Consumer<Message<String>> consume2() {
    return message -> {
        Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
        LOGGER.info("New message2 received: '{}'", message);
        checkpointer.success()
                .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                .doOnError(e -> LOGGER.error("Error found", e))
                .block();
    };

}

Resurstilldelning

Service Bus Binder stöder etablering av kö, ämne och prenumeration. Användarna kan använda följande egenskaper för att aktivera etablering.

spring:
  cloud:
    azure:
      credential:
        tenant-id: <tenant>
      profile:
        subscription-id: ${AZURE_SUBSCRIPTION_ID}
      servicebus:
        resource:
          resource-group: ${AZURE_SERVICEBUS_RESOURCE_GROUP}
    stream:
      servicebus:
        bindings:
          <binding-name>:
            consumer:
              entity-type: ${SERVICEBUS_CONSUMER_ENTITY_TYPE}

Obs!

De värden som tillåts för tenant-id är: common, organizations, consumerseller klientorganisations-ID. Mer information om de här värdena finns i avsnittet Fel slutpunkt användes (personliga konton och organisationskonton) i Fel AADSTS50020 – Användarkonto från identitetsleverantören finns inte i klientorganisationen. Information om hur du konverterar din app för en klientorganisation finns i Konvertera enklientapp till flera klientorganisationer på Microsoft Entra-ID.

Anpassa Service Bus-klientegenskaper

Utvecklare kan använda AzureServiceClientBuilderCustomizer för att anpassa Service Bus-klientegenskaper. I följande exempel anpassas egenskapen sessionIdleTimeout i ServiceBusClientBuilder:

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

Exempel

Mer information finns i lagringsplatsen azure-spring-boot-samples på GitHub.

  • Last updated on