Spring Cloud Azure 對 Spring Integration 的支援

適用於 Azure 的 Spring Integration Extension 為 Azure SDK for Java 提供的各種服務提供 Spring Integration 配接器。 我們提供這些 Azure 服務的 Spring Integration 支援：事件中樞、服務總線、記憶體佇列。 以下是支援的配接器清單：

• spring-cloud-azure-starter-integration-eventhubs - 如需詳細資訊，請參閱 Spring Integration with Azure 事件中樞
• spring-cloud-azure-starter-integration-servicebus - 如需詳細資訊，請參閱 Spring Integration with Azure 服務匯流排
• spring-cloud-azure-starter-integration-storage-queue - 如需詳細資訊，請參閱 Spring Integration with Azure 儲存體 Queue

Spring 與 Azure 事件中樞整合

重要概念

Azure 事件中樞是巨量數據串流平臺和事件擷取服務。 它可以每秒接收和處理數百萬個事件。 傳送至事件中樞的數據可以使用任何即時分析提供者或批處理/記憶體配接器來轉換和儲存。

Spring Integration 可在 Spring 型應用程式中啟用輕量型傳訊，並支援透過宣告式配接器與外部系統整合。 這些配接器在 Spring 對遠端、訊息傳遞和排程的支援之上，提供了更高層級的抽象。 Spring Integration for Event Hubs 擴充專案為 Azure 事件中樞提供輸入與輸出通道配接器及閘道。

取用者群組

事件中樞提供與 Apache Kafka 類似的取用者群組支援，但邏輯稍有不同。 雖然 Kafka 會將所有已提交的位移儲存在訊息代理中，但您必須手動儲存正在處理的 Event Hubs 訊息位移。 事件中樞 SDK 提供函式，以將這類位移儲存在 Azure 記憶體內。

資料分割支援

事件中樞提供與 Kafka 類似的實體分割區概念。 但不同於 Kafka 在取用者與分割區之間的自動重新平衡，Event Hubs 採用一種預先搶占模式。 儲存體帳戶可作為租約，用來判斷哪個分割區由哪個取用者擁有。 當新的消費者啟動時，它會嘗試從負載最重的消費者中竊取一些分割區，以達成負載平衡。

若要指定負載平衡策略，開發人員可以使用 EventHubsContainerProperties 來進行設定。 如需設定 的範例，請參閱以下章節。

批次取用端支援

EventHubsInboundChannelAdapter 支援批次消費模式。 若要啟用，用戶可以在建構 ListenerMode.BATCH 實例時，將接聽程式模式指定為 EventHubsInboundChannelAdapter。 啟用時，將會接收一則 訊息，其承載內容為批次事件清單，並將其傳遞至下游通道。 每個訊息標頭也會轉換成清單，其中內容是從每個事件剖析的相關聯標頭值。 對於分割區識別碼、checkpointer 和最後排入佇列屬性的共用標頭，若整個事件批次共用相同的值，則會顯示為單一值。 如需詳細資訊，請參閱 事件中樞訊息標頭 一節。

批次取用程式的檢查點功能支援兩種模式：BATCH 和 MANUAL。 BATCH 模式是一種自動建立檢查點的模式，會在收到事件後，為整個事件批次一併建立檢查點。 MANUAL 模式用於讓使用者為事件建立檢查點。 使用時，Checkpointer 會傳遞至訊息標頭，而且使用者可以使用它來執行檢查點。

批次取用原則可透過 max-size 和 max-wait-time 的屬性指定，其中 max-size 為必要屬性，而 max-wait-time 為選用屬性。 若要指定批次取用策略，開發人員可以使用 EventHubsContainerProperties 來進行設定。 如需設定 的範例，請參閱 以下章節。

相依性設定

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

配置

此入門套件提供以下 3 部分的組態選項：

聯機組態屬性

本節包含用來連線到 Azure 事件中樞的組態選項。

spring-cloud-azure-starter-integration-eventhubs 的連線可設定屬性：

檢查點組態屬性

本節包含記憶體 Blob 服務的組態選項，用於保存分割區擁有權和檢查點資訊。

spring-cloud-azure-starter-integration-eventhubs 的檢查點可設定屬性：

常見的 Azure 服務 SDK 組態選項也可以針對記憶體 Blob 檢查點存放區進行設定。 支援的組態選項已於 Spring Cloud Azure 組態中介紹，且可使用統一前綴 spring.cloud.azure. 或 spring.cloud.azure.eventhubs.processor.checkpoint-store. 前綴進行設定。

事件中樞處理器組態屬性

EventHubsInboundChannelAdapter 會使用 EventProcessorClient 從事件中樞取用訊息，來設定 EventProcessorClient的整體屬性，開發人員可以使用 EventHubsContainerProperties 進行設定。 請參閱下一節 如何使用 EventHubsInboundChannelAdapter。

基本用法

將訊息傳送至 Azure 事件中樞

1. 填入認證組態選項。

   • 針對認證作為連接字串，請在 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}
     

     注意

     Microsoft 建議您使用最安全的可用驗證流程。 此程式中所述的驗證流程，例如資料庫、快取、傳訊或 AI 服務，在應用程式中需要高度的信任，而且不會在其他流程中帶來風險。 只有在更安全的選項（例如使用受控識別進行無密碼或無金鑰連線）不可行時，才使用此流程。 針對本機計算機作業，偏好使用無密碼或無密鑰連線的使用者身分識別。

   • 針對認證作為受控識別，請在 application.yml 檔案中設定下列屬性：

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

   • 若使用服務主體認證，請在 application.yml 檔案中設定下列屬性：

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

1. 使用 EventHubsTemplate Bean 建立 DefaultMessageHandler，以將訊息傳送至 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. 透過訊息通道建立具有上述訊息處理程式的訊息網關係結。

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

3. 使用閘道傳送訊息。

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

從 Azure 事件中樞接收訊息

1. 填入認證組態選項。

2. 建立訊息通道的 Bean 做為輸入通道。

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

3. 使用 EventHubsMessageListenerContainer bean 建立 EventHubsInboundChannelAdapter，以接收來自事件中樞的訊息。

   @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. 透過之前建立的訊息通道，使用 EventHubsInboundChannelAdapter 建立訊息接收者系結。

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

設定 EventHubsMessageConverter 以自定義 objectMapper

EventHubsMessageConverter 被設計為可配置的 Bean，讓使用者能夠自訂 ObjectMapper。

批次取用端支援

若要從 Event Hubs 以批次方式取用訊息，作法與上述範例類似；此外，使用者應在 EventHubsInboundChannelAdapter 中設定與批次取用相關的組態選項。

建立 EventHubsInboundChannelAdapter 時，接聽模式應設為 BATCH。 建立 EventHubsMessageListenerContainer的 bean 時，請將檢查點模式設定為 MANUAL 或 BATCH，並視需要設定批次選項。

@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 訊息標頭

下表說明事件中樞訊息屬性如何對應至 Spring 訊息標頭。 針對 Azure 事件中樞，訊息稱為 event。

在記錄接聽器模式中，Event Hubs 訊息 / 事件屬性與 Spring 訊息標頭之間的對應：

使用者可以剖析訊息標頭，以取得每個事件的相關信息。 若要設定事件的訊息標頭，所有自定義標頭都會放置為事件的應用程式屬性，其中標頭會設定為屬性索引鍵。 從事件中樞接收事件時，所有應用程式屬性都會轉換成訊息標頭。

啟用批次取用者模式時，批次訊息的特定標頭如下所列，其中包含每個個別 Event Hubs 事件的值清單。

在批次接聽器模式下，Event Hubs 訊息/事件屬性與 Spring 訊息標頭之間的對應關係：

樣品

欲了解更多資訊，請參閱 azure-spring-boot-samples GitHub 上的倉庫。

Spring 與 Azure 服務總線整合

重要概念

Spring Integration 可在 Spring 型應用程式中啟用輕量型傳訊，並支援透過宣告式配接器與外部系統整合。

Azure 服務總線延伸模組專案的 Spring Integration 提供 Azure 服務總線的輸入和輸出通道配接器。

相依性設定

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

配置

此入門範本提供下列兩部分的設定選項：

聯機組態屬性

本節包含用來連線到 Azure 服務總線的組態選項。

spring-cloud-azure-starter-integration-servicebus 的連線可設定屬性：

服務總線處理器組態屬性

ServiceBusInboundChannelAdapter 會使用 ServiceBusProcessorClient 來取用訊息，來設定 ServiceBusProcessorClient的整體屬性，開發人員可以使用 ServiceBusContainerProperties 來進行設定。 請參閱下一節 如何使用 ServiceBusInboundChannelAdapter。

基本用法

將訊息傳送至 Azure 服務總線

1. 填入認證組態選項。

   • 針對認證作為連接字串，請在 application.yml 檔案中設定下列屬性：

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

     注意

     Microsoft 建議您使用最安全的可用驗證流程。 此程式中所述的驗證流程，例如資料庫、快取、傳訊或 AI 服務，在應用程式中需要高度的信任，而且不會在其他流程中帶來風險。 只有在更安全的選項（例如使用受控識別進行無密碼或無金鑰連線）不可行時，才使用此流程。 針對本機計算機作業，偏好使用無密碼或無密鑰連線的使用者身分識別。

   • 針對認證作為受控識別，請在 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}
     

• 若使用服務主體認證，請在 application.yml 檔案中設定下列屬性：

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

1. 使用 ServiceBusTemplate bean 建立 DefaultMessageHandler 以將訊息傳送至 服務匯流排，並為 ServiceBusTemplate 設定實體類型。 此範例會採用服務總線佇列作為範例。

   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. 透過訊息通道建立具有上述訊息處理程式的訊息網關係結。

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

3. 使用閘道傳送訊息。

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

從 Azure 服務總線接收訊息

1. 填入認證組態選項。

2. 建立訊息通道的 Bean 做為輸入通道。

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

3. 使用 ServiceBusMessageListenerContainer Bean 建立 ServiceBusInboundChannelAdapter，以接收來自 服務匯流排 的訊息。 此範例會採用服務總線佇列作為範例。

   @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. 透過我們先前建立的訊息通道，建立具有 ServiceBusInboundChannelAdapter 的訊息接收者系結。

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

設定 ServiceBusMessageConverter 以自定義 objectMapper

ServiceBusMessageConverter 被設計為可配置的 Bean，讓使用者能夠自訂 ObjectMapper。

服務總線訊息標頭

對於一些可以對應至多個 Spring 標頭常數的服務總線標頭，會列出不同 Spring 標頭的優先順序。

服務總線標頭與 Spring 標頭之間的對應：

分割區索引鍵支援

此入門支援 服務總線分割，方法是允許在訊息標頭中設定分割區索引鍵和會話標識符。 本節將介紹如何為訊息設定分割區索引鍵。

建議：使用 ServiceBusMessageHeaders.PARTITION_KEY 作為標頭的鍵。

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(ServiceBusMessageHeaders.PARTITION_KEY, "Customize partition key")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

不建議，但目前仍支援將 AzureHeaders.PARTITION_KEY 作為標頭的鍵。

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(AzureHeaders.PARTITION_KEY, "Customize partition key")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

會話支援

此範例示範如何在應用程式中手動設定訊息的會話標識碼。

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

自訂服務總線客戶端屬性

開發人員可以使用 AzureServiceClientBuilderCustomizer 來自定義服務總線用戶端屬性。 下列範例會自訂 sessionIdleTimeout中的 ServiceBusClientBuilder 屬性：

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

樣品

欲了解更多資訊，請參閱 azure-spring-boot-samples GitHub 上的倉庫。

Spring 與 Azure 記憶體佇列整合

重要概念

Azure 佇列記憶體是用來儲存大量訊息的服務。 您可以使用 HTTP 或 HTTPS 透過已驗證的呼叫，從世界各地存取訊息。 佇列訊息的大小最多可達 64 KB。 佇列可能包含數百萬則訊息，最多可達記憶體帳戶的總容量限制。 佇列通常用於建立待處理工作的積壓，供非同步處理。

相依性設定

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

配置

此入門提供下列組態選項：

聯機組態屬性

本節包含用來連線到 Azure 記憶體佇列的組態選項。

spring-cloud-azure-starter-integration-storage-queue 的連線可設定屬性：

基本用法

將訊息傳送至 Azure 記憶體佇列

1. 填入認證組態選項。

   • 針對認證作為連接字串，請在 application.yml 檔案中設定下列屬性：

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

     注意

     Microsoft 建議您使用最安全的可用驗證流程。 此程式中所述的驗證流程，例如資料庫、快取、傳訊或 AI 服務，在應用程式中需要高度的信任，而且不會在其他流程中帶來風險。 只有在更安全的選項（例如使用受控識別進行無密碼或無金鑰連線）不可行時，才使用此流程。 針對本機計算機作業，偏好使用無密碼或無密鑰連線的使用者身分識別。

   • 針對認證作為受控識別，請在 application.yml 檔案中設定下列屬性：

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

• 若使用服務主體認證，請在 application.yml 檔案中設定下列屬性：

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

1. 使用 StorageQueueTemplate Bean 建立 DefaultMessageHandler，以將訊息傳送至儲存體佇列。

   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. 透過訊息通道，使用上述訊息處理程式建立訊息網關係結。

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

3. 使用閘道傳送訊息。

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

從 Azure 記憶體佇列接收訊息

1. 填入認證組態選項。

2. 建立訊息通道的 Bean 做為輸入通道。

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

3. 使用 StorageQueueTemplate Bean 建立 StorageQueueMessageSource，以接收來自儲存體佇列的訊息。

   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. 透過我們先前建立的訊息通道，使用上一個步驟中建立的 StorageQueueMessageSource 建立訊息接收器繫結。

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

樣品

欲了解更多資訊，請參閱 azure-spring-boot-samples GitHub 上的倉庫。