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

Spring Integration Extension för Azure tillhandahåller Spring Integration-kort för de olika tjänster som tillhandahålls av Azure SDK för Java. Vi tillhandahåller Spring Integration-stöd för dessa Azure-tjänster: Event Hubs, Service Bus, Storage Queue. Följande är en lista över kort som stöds:

• spring-cloud-azure-starter-integration-eventhubs – mer information finns i Spring Integration med Azure Event Hubs
• spring-cloud-azure-starter-integration-servicebus – mer information finns i Spring Integration med Azure Service Bus
• spring-cloud-azure-starter-integration-storage-queue – mer information finns i Spring-integrering med Azure Storage Queue

Spring-integrering med Azure Event Hubs

Viktiga begrepp

Azure Event Hubs är en stordataströmningsplattform och händelseinmatningstjänst. Den kan ta emot och bearbeta miljontals händelser per sekund. Data som skickas till en händelsehubb kan transformeras och lagras med hjälp av valfri realtidsanalysprovider eller batchbearbetning/lagringskort.

Spring Integration möjliggör enklare meddelanden i Spring-baserade program och stöder integrering med externa system via deklarativa kort. Dessa kort ger en högre abstraktionsnivå jämfört med Spring-stöd för fjärrkommunikation, meddelanden och schemaläggning. Spring Integration for Event Hubs-tilläggsprojektet tillhandahåller inkommande och utgående kanalkort och gatewayer för Azure Event Hubs.

Konsumentgrupp

Event Hubs har liknande stöd för konsumentgruppen som Apache Kafka, men med lite annan logik. Medan Kafka lagrar alla incheckade förskjutningar i asynkron meddelandekö måste du lagra förskjutningar av Event Hubs-meddelanden som bearbetas manuellt. Event Hubs SDK tillhandahåller funktionen för att lagra sådana förskjutningar 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 ett lån för att avgöra vilken partition som ägs av vilken konsument. När en ny konsument startar försöker den stjäla vissa partitioner från de flesta tunga användare för att uppnå arbetsbelastningsutjämningen.

För att ange belastningsutjämningsstrategin kan utvecklare använda EventHubsContainerProperties för konfigurationen. I följande avsnitt ett exempel på hur du konfigurerar EventHubsContainerProperties.

Stöd för Batch-konsument

EventHubsInboundChannelAdapter stöder batchförbrukningsläget. För att aktivera det kan användarna ange lyssnarläget som ListenerMode.BATCH när de skapar en EventHubsInboundChannelAdapter instans. När det är aktiverat tas ett meddelande som nyttolasten är en lista över batchade händelser emot och skickas till nedströmskanalen. Varje meddelanderubrik konverteras också som en lista, där innehållet är det associerade rubrikvärdet som parsas från varje händelse. För de gemensamma huvudena för partitions-ID, checkpointer och de senaste egenskaperna visas de som ett enda värde för hela batchen med händelser som delar samma. Mer information finns i avsnittet Event Hubs-meddelandehuvuden.

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 de tas emot. MANUAL läge är att kontrollera händelserna av användare. När den används skickas Checkpointer- till meddelanderubriken och användarna kan använda den för att göra kontrollpunkter.

Principen för batchförbrukning kan anges av egenskaperna för max-size och max-wait-time, där max-size är en nödvändig egenskap medan max-wait-time är valfritt. Utvecklare kan använda EventHubsContainerProperties för konfigurationen för att ange den batchkrävande strategin. I följande avsnitt ett exempel på hur du konfigurerar EventHubsContainerProperties.

Beroendekonfiguration

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

Konfiguration

Den här startprogrammet innehåller följande tre delar av konfigurationsalternativen:

Egenskaper för anslutningskonfiguration

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

Anslutningskonfigurerbara egenskaper för spring-cloud-azure-starter-integration-eventhubs:

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.

Kontrollpunktskonfigurerbara egenskaper för spring-cloud-azure-starter-integration-eventhubs:

Vanliga konfigurationsalternativ för Azure Service SDK kan även konfigureras för Lagringsblobkontrollpunktsarkiv. 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 Event Hub-processor

EventHubsInboundChannelAdapter använder EventProcessorClient för att använda meddelanden från en händelsehubb för att konfigurera de övergripande egenskaperna för en EventProcessorClientkan utvecklare använda EventHubsContainerProperties för konfigurationen. Se följande avsnitt om hur du arbetar med EventHubsInboundChannelAdapter.

Grundläggande användning

Skicka meddelanden till Azure Event Hubs

1. Fyll i konfigurationsalternativen för autentiseringsuppgifter.

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

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

     Not

     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 som hanterade identiteter konfigurerar du följande egenskaper i din application.yml-fil:

     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}
     

   • För autentiseringsuppgifter som tjänstens huvudnamn 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: ${AZURE_EVENT_HUBS_NAMESPACE}
             processor:
               checkpoint-store:
                 container-name: ${CONTAINER_NAME}
                 account-name: ${ACCOUNT_NAME}
     

1. Skapa DefaultMessageHandler med den EventHubsTemplate bönan för att skicka meddelanden till Event Hubs.

   class Demo {
       private static final String OUTPUT_CHANNEL = "output";
       private static final String EVENTHUB_NAME = "eh1";
   
       @Bean
       @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
       public MessageHandler messageSender(EventHubsTemplate eventHubsTemplate) {
           DefaultMessageHandler handler = new DefaultMessageHandler(EVENTHUB_NAME, eventHubsTemplate);
           handler.setSendCallback(new ListenableFutureCallback<Void>() {
               @Override
               public void onSuccess(Void result) {
                   LOGGER.info("Message was sent successfully.");
               }
               @Override
               public void onFailure(Throwable ex) {
                   LOGGER.error("There was an error sending the message.", ex);
               }
           });
           return handler;
       }
   }
   

2. Skapa en meddelandegatewaybindning med meddelandehanteraren ovan via en meddelandekanal.

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

3. Skicka meddelanden med hjälp av gatewayen.

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

Ta emot meddelanden från Azure Event Hubs

1. Fyll i konfigurationsalternativen för autentiseringsuppgifter.

2. Skapa en böna med meddelandekanalen som indatakanal.

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

3. Skapa EventHubsInboundChannelAdapter med EventHubsMessageListenerContainer böna för att ta emot meddelanden från Event Hubs.

   @Configuration
   class Demo {
       private static final String INPUT_CHANNEL = "input";
       private static final String EVENTHUB_NAME = "eh1";
       private static final String CONSUMER_GROUP = "$Default";
   
       @Bean
       public EventHubsInboundChannelAdapter messageChannelAdapter(
               @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
               EventHubsMessageListenerContainer listenerContainer) {
           EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(processorContainer);
           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. Skapa en meddelandemottagarebindning med EventHubsInboundChannelAdapter via meddelandekanalen som skapades tidigare.

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

Konfigurera EventHubsMessageConverter för att anpassa objectMapper

EventHubsMessageConverter görs som en konfigurerbar böna så att användarna kan anpassa ObjectMapper.

Stöd för Batch-konsument

Att använda meddelanden från Event Hubs i batchar liknar exemplet ovan, förutom att användarna bör ange de batchkrävande relaterade konfigurationsalternativen för EventHubsInboundChannelAdapter.

När du skapar EventHubsInboundChannelAdapterska lyssnarläget anges som BATCH. När du skapar en böna för EventHubsMessageListenerContaineranger du kontrollpunktsläget som antingen MANUAL eller BATCH, och batchalternativen kan konfigureras efter behov.

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

Meddelandehuvuden för Event Hubs

I följande tabell visas hur event hubs-meddelandeegenskaper mappas till Spring-meddelandehuvuden. För Azure Event Hubs anropas meddelandet som event.

Mappning mellan Event Hubs-meddelande/händelseegenskaper och Spring Message-huvuden i inspelningslyssningsläge:

Användare kan parsa meddelanderubrikerna för relaterad information om varje händelse. Om du vill ange ett meddelandehuvud för händelsen placeras alla anpassade rubriker som en programegenskap för en händelse, där rubriken anges som egenskapsnyckel. När händelser tas emot från Event Hubs konverteras alla programegenskaper till meddelandehuvudet.

När batch-konsumentläget är aktiverat visas de specifika rubrikerna för batchmeddelanden på följande sätt, som innehåller en lista med värden från varje händelse i Händelsehubbar.

Mappning mellan Event Hubs-meddelande/händelseegenskaper och Spring Message-huvuden i Batch-lyssnarläge:

Prover

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

Spring-integrering med Azure Service Bus

Viktiga begrepp

Spring Integration möjliggör enklare meddelanden i Spring-baserade program och stöder integrering med externa system via deklarativa kort.

Spring Integration for Azure Service Bus-tilläggsprojektet tillhandahåller inkommande och utgående kanalkort för Azure Service Bus.

Beroendekonfiguration

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

Konfiguration

Den här startprogrammet innehåller följande två delar av konfigurationsalternativen:

Egenskaper för anslutningskonfiguration

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

Anslutningskonfigurerbara egenskaper för spring-cloud-azure-starter-integration-servicebus:

Konfigurationsegenskaper för Service Bus-processor

ServiceBusInboundChannelAdapter använder ServiceBusProcessorClient för att använda meddelanden för att konfigurera de övergripande egenskaperna för en ServiceBusProcessorClient, kan utvecklare använda ServiceBusContainerProperties för konfigurationen. Se följande avsnitt om hur du arbetar med ServiceBusInboundChannelAdapter.

Grundläggande användning

Skicka meddelanden till Azure Service Bus

1. Fyll i konfigurationsalternativen för autentiseringsuppgifter.

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

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

     Not

     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 som hanterade identiteter konfigurerar du följande egenskaper i din application.yml-fil:

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

• För autentiseringsuppgifter som tjänstens huvudnamn 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: ${AZURE_SERVICE_BUS_NAMESPACE}
  

1. Skapa DefaultMessageHandler med den ServiceBusTemplate bönan för att skicka meddelanden till Service Bus, ange entitetstypen för ServiceBusTemplate. Det här exemplet tar Service Bus Queue som exempel.

   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.info("There was an error sending the message.");
               }
           });
   
           return handler;
       }
   }
   

2. Skapa en meddelandegatewaybindning med meddelandehanteraren ovan via en meddelandekanal.

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

3. Skicka meddelanden med hjälp av gatewayen.

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

Ta emot meddelanden från Azure Service Bus

1. Fyll i konfigurationsalternativen för autentiseringsuppgifter.

2. Skapa en böna med meddelandekanalen som indatakanal.

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

3. Skapa ServiceBusInboundChannelAdapter med den ServiceBusMessageListenerContainer bönan för att ta emot meddelanden till Service Bus. Det här exemplet tar Service Bus Queue som exempel.

   @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. Skapa en meddelandemottagarebindning med ServiceBusInboundChannelAdapter via meddelandekanalen som vi skapade tidigare.

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

Konfigurera ServiceBusMessageConverter för att anpassa objectMapper

ServiceBusMessageConverter görs som en konfigurerbar böna så att användarna kan anpassa ObjectMapper.

Service Bus-meddelandehuvuden

För vissa Service Bus-huvuden som kan mappas till flera Spring-huvudkonstanter visas prioriteten för olika Spring-huvuden.

Mappning mellan Service Bus-huvuden och Spring-huvuden:

Stöd för partitionsnyckel

Den här startprogrammet 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.

Rekommenderas: Använd ServiceBusMessageHeaders.PARTITION_KEY som nyckel i rubriken.

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

Rekommenderas inte men stöds för närvarande: AzureHeaders.PARTITION_KEY som nyckeln i rubriken.

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

Sessionsstöd

Det här exemplet visar hur du manuellt anger sessions-ID för ett meddelande i programmet.

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

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

Prover

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

Spring-integrering med Azure Storage Queue

Viktiga begrepp

Azure Queue Storage är en tjänst för att lagra ett stort antal meddelanden. Du kommer åt meddelanden var som helst i världen via autentiserade anrop med HTTP eller HTTPS. Ett kömeddelande kan vara upp till 64 KB stort. En kö kan innehålla miljontals meddelanden, upp till den totala kapacitetsgränsen för ett lagringskonto. Köer används ofta för att skapa en kvarvarande arbetslogg för att bearbeta asynkront.

Beroendekonfiguration

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

Konfiguration

Den här startprogrammet innehåller följande konfigurationsalternativ:

Egenskaper för anslutningskonfiguration

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

Anslutningskonfigurerbara egenskaper för spring-cloud-azure-starter-integration-storage-queue:

Grundläggande användning

Skicka meddelanden till Azure Storage Queue

1. Fyll i konfigurationsalternativen för autentiseringsuppgifter.

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

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

     Not

     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 som hanterade identiteter konfigurerar du följande egenskaper i din application.yml-fil:

     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}
     

• För autentiseringsuppgifter som tjänstens huvudnamn 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>
        storage:
          queue:
            account-name: ${AZURE_STORAGE_QUEUE_ACCOUNT_NAME}
  

1. Skapa DefaultMessageHandler med den StorageQueueTemplate bönan för att skicka meddelanden till Lagringskö.

   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.info("There was an error sending the message.");
               }
           });
           return handler;
       }
   }
   

2. Skapa en meddelandegatewaybindning med meddelandehanteraren ovan via en meddelandekanal.

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

3. Skicka meddelanden med hjälp av gatewayen.

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

Ta emot meddelanden från Azure Storage Queue

1. Fyll i konfigurationsalternativen för autentiseringsuppgifter.

2. Skapa en böna med meddelandekanalen som indatakanal.

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

3. Skapa StorageQueueMessageSource med den StorageQueueTemplate bönan för att ta emot meddelanden till Lagringskö.

   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. Skapa en meddelandemottagarebindning med StorageQueueMessageSource som skapades i det sista steget via meddelandekanalen som vi skapade tidigare.

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

Prover

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