具有 Azure 事件中樞的 Spring Cloud Stream

發行項
04/06/2024

本教學課程示範如何在 Spring Boot 應用程式中使用 Azure 事件中樞和 Spring Cloud Stream Binder Eventhubs 來傳送和接收訊息。

必要條件

Azure 訂用帳戶 - 建立免費帳戶。
Java Development Kit （JDK）第 8 版或更高版本。
Apache Maven 3.2 版或更高版本。
cURL 或類似的 HTTP 公用程式來測試功能。
Azure 事件中樞。如果您沒有事件中樞，請使用 Azure 入口網站建立事件中樞。
事件中樞檢查點 Azure 儲存體帳戶。如果您沒有記憶體帳戶，請建立記憶體帳戶。
Spring Boot 應用程式。如果您沒有這個應用程式，請使用 Spring Initializr 來建立 Maven 專案。請務必選取 Maven 專案，然後在 [相依性] 底下新增 Spring Web 和 Azure 支援相依性，然後選取 [Java 第 8 版] 或更新版本。

注意

若要授與帳戶對資源的存取權，請在 Azure 事件中樞中，將和 Azure Event Hubs Data Sender 角色指派Azure Event Hubs Data Receiver給您目前使用的 Microsoft Entra 帳戶。然後，在 Azure 儲存體帳戶中，將角色指派Storage Blob Data Contributor給您目前使用的 Microsoft Entra 帳戶。如需授與存取角色的詳細資訊，請參閱使用 Azure 入口網站指派 Azure 角色，並使用 Microsoft Entra ID 授權事件中樞資源的存取權。

重要

需要 Spring Boot 2.5 版或更高版本，才能完成本教學課程中的步驟。

從 Azure 事件中樞傳送和接收訊息

透過 Azure 儲存體帳戶和 Azure 事件中樞，您可以使用 Spring Cloud Azure Stream Binder 事件中樞來傳送和接收訊息。

若要安裝 Spring Cloud Azure Stream Binder 事件中樞模組，請將下列相依性新增至 您的pom.xml 檔案：

Spring Cloud Azure 材料帳單（BOM）：
```
<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>com.azure.spring</groupId>
      <artifactId>spring-cloud-azure-dependencies</artifactId>
      <version>5.18.0</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>
```
注意

如果您使用 Spring Boot 2.x，請務必將 spring-cloud-azure-dependencies 版本設定為 4.19.0。此材料帳單（BOM）應該在<dependencyManagement>pom.xml檔案的區段中設定。這可確保所有 Spring Cloud Azure 相依性都使用相同的版本。如需此 BOM 所用版本的詳細資訊，請參閱應該使用哪個版本的 Spring Cloud Azure。

Spring Cloud Azure Stream Binder 事件中樞成品：

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

編碼應用程式

使用下列步驟來設定應用程式，以使用 Azure 事件中樞產生和取用訊息。

將下列屬性新增至 application.properties 檔案，以設定事件中樞認證。

 spring.cloud.azure.eventhubs.namespace=${AZURE_EVENTHUBS_NAMESPACE}
 spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=${AZURE_STORAGE_ACCOUNT_NAME}
 spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=${AZURE_STORAGE_CONTAINER_NAME}
 spring.cloud.stream.bindings.consume-in-0.destination=${AZURE_EVENTHUB_NAME}
 spring.cloud.stream.bindings.consume-in-0.group=${AZURE_EVENTHUB_CONSUMER_GROUP}
 spring.cloud.stream.bindings.supply-out-0.destination=${AZURE_EVENTHUB_NAME}
 spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode=MANUAL
 spring.cloud.function.definition=consume;supply;
 spring.cloud.stream.poller.initial-delay=0
 spring.cloud.stream.poller.fixed-delay=1000

下表描述群組態中的欄位：

欄位	描述
`spring.cloud.azure.eventhubs.namespace`	指定您在事件中樞從 Azure 入口網站取得的命名空間。
`spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name`	指定您在本教學課程中建立的記憶體帳戶。
`spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name`	指定記憶體帳戶的容器。
`spring.cloud.stream.bindings.consume-in-0.destination`	指定您在本教學課程中使用的事件中樞。
`spring.cloud.stream.bindings.consume-in-0.group`	在事件中樞實例中指定取用者群組。
`spring.cloud.stream.bindings.supply-out-0.destination`	指定您在本教學課程中使用的相同事件中樞。
`spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode`	指定 `MANUAL`。
`spring.cloud.function.definition`	指定要系結至系結所公開之外部目的地的功能豆。
`spring.cloud.stream.poller.initial-delay`	指定定期觸發程式的初始延遲。預設值為 0。
`spring.cloud.stream.poller.fixed-delay`	指定預設輪詢器以毫秒為單位的固定延遲。預設值為 1000 L。

編輯啟動類別檔案以顯示下列內容。

import com.azure.spring.messaging.checkpoint.Checkpointer;
import com.azure.spring.messaging.eventhubs.support.EventHubsHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.util.function.Consumer;
import java.util.function.Supplier;
import static com.azure.spring.messaging.AzureHeaders.CHECKPOINTER;

@SpringBootApplication
public class EventHubBinderApplication implements CommandLineRunner {

    private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderApplication.class);
    private static final Sinks.Many<Message<String>> many = Sinks.many().unicast().onBackpressureBuffer();

    public static void main(String[] args) {
        SpringApplication.run(EventHubBinderApplication.class, args);
    }

    @Bean
    public Supplier<Flux<Message<String>>> supply() {
        return ()->many.asFlux()
                       .doOnNext(m->LOGGER.info("Manually sending message {}", m))
                       .doOnError(t->LOGGER.error("Error encountered", t));
    }

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

    @Override
    public void run(String... args) {
        LOGGER.info("Going to add message {} to sendMessage.", "Hello World");
        many.emitNext(MessageBuilder.withPayload("Hello World").build(), Sinks.EmitFailureHandler.FAIL_FAST);
    }

}

提示

在本教學課程中，組態或程式代碼中沒有任何驗證作業。不過，連線到 Azure 服務需要驗證。若要完成驗證，您需要使用 Azure Identity。 Spring Cloud Azure 使用 DefaultAzureCredential，Azure 身分識別連結庫會提供它來協助您取得認證，而不需要變更任何程序代碼。

DefaultAzureCredential 支援多種驗證方法，並在執行階段判斷應使用的方法。這種方法可讓您的應用程式在不同的環境中使用不同的驗證方法（例如本機和生產環境），而不需要實作環境特定的程序代碼。如需詳細資訊，請參閱 DefaultAzureCredential。

若要在本機開發環境中完成驗證，您可以使用 Azure CLI、Visual Studio Code、PowerShell 或其他方法。如需詳細資訊，請參閱 Java 開發環境中的 Azure 驗證。若要在 Azure 裝載環境中完成驗證，建議您使用使用者指派的受控識別。如需詳細資訊，請參閱什麼是 Azure 資源受控識別？

啟動應用程式。這類訊息將會張貼在您的應用程式記錄檔中，如下列範例輸出所示：

New message received: 'Hello World', partition key: 107207233, sequence number: 458, offset: 94256, enqueued time: 2023-02-17T08:27:59.641Z
Message 'Hello World!' successfully checkpointed

部署至 Azure Spring Apps

現在您已在本機執行 Spring Boot 應用程式，現在可以將其移至生產環境。 Azure Spring Apps 可讓您輕鬆地將 Spring Boot 應用程式部署至 Azure，而不需要變更任何程式代碼。服務會管理 Spring 應用程式的基礎結構，讓開發人員可以專注於處理程式碼。 Azure Spring 應用程式提供生命週期管理，使用全方位的監視和診斷、組態管理、服務探索、持續整合與持續傳遞的整合、藍綠部署等等。若要將應用程式部署至 Azure Spring Apps，請參閱將第一個應用程式部署至 Azure Spring Apps。

下一步

適用於 Spring 開發人員的 Azure Spring Cloud Azure Stream Binder 事件中樞範例

共用方式為

具有 Azure 事件中樞的 Spring Cloud Stream

必要條件

從 Azure 事件中樞傳送和接收訊息

編碼應用程式

部署至 Azure Spring Apps

下一步

意見反應

其他資源

共用方式為

具有 Azure 事件中樞 的 Spring Cloud Stream

必要條件

從 Azure 事件中樞 傳送和接收訊息

編碼應用程式

部署至 Azure Spring Apps

下一步

意見反應

其他資源

具有 Azure 事件中樞的 Spring Cloud Stream

從 Azure 事件中樞傳送和接收訊息