SpringアプリケーションでのEvent Hubsを使用する

この記事では、Spring Framework でビルドされたJava アプリケーションでAzure Event Hubsを使用する方法について説明します。

Azure Event Hubs は、ビッグ データ ストリーミング プラットフォームおよびイベント インジェスト サービスです。 1 秒間に何百万ものイベントを受信して処理することができます。 イベント ハブに送信されたデータは、任意のリアルタイム分析プロバイダーやバッチ処理/ストレージ アダプターを使用して、変換および保存できます。

Spring Cloud Azure には、Spring フレームワークを使用して Event Hubs との間でメッセージを送受信するためのさまざまなモジュールが用意されています。

次のモジュールは個別に使用することも、さまざまな用途で組み合わせて使用することもできます。

• Spring Cloud Azure Event Hubs Starter を使用すると、Spring Boot 機能を備えた Event Hubs Java SDK クライアント ライブラリでメッセージを送受信できます。

• Spring Messaging Azure Event Hubs を使用すると、Spring Messaging API を介して Event Hubs と対話できます。

• Spring Integration Azure Event Hubs を使用すると、Spring Integration Message Channels を Event Hubs に接続できます。

• Spring Cloud Azure Stream Event Hubs Binder を使用すると、Spring Cloud Stream アプリケーションのメッセージング ミドルウェアとして Event Hubs を使用できます。

• Spring Kafka with Azure Event Hubs を使用すると、Spring Kafka を使用して Event Hubs との間でメッセージを送受信できます。

• Spring Cloud Stream Kafka Binder with Azure Event Hubs を使用すると、Event Hubs を使用して Spring Cloud Stream Kafka Binder 経由でメッセージを送受信できます。

前提条件

• Azure サブスクリプション - 無料で作成。

• Java Development Kit (JDK) バージョン 8 以降。

• Apache Maven。

• Azure CLI。

• Azure Event Hubs インスタンス。 詳細については、「Quickstart: Azure ポータルを使用してイベント ハブを作成するを参照してください。

• Event Hubs チェックポイントのAzure Storage アカウント。 詳しくは、「ストレージ アカウントの作成」をご覧ください。

• Spring Boot アプリケーション。 ない場合は、Spring InitializrでMavenプロジェクトを作成します。 必ず Maven Project を選択し、Dependencies の下に Spring Web 依存関係を追加してから、バージョン 8 以降Java選択します。

注

アカウントにリソースへのアクセス権を付与するには、Azure Event Hubsで、現在使用しているMicrosoft Entra アカウントに Azure Event Hubs Data Receiver ロールと Azure Event Hubs Data Sender ロールを割り当てます。 次に、Azure Storage アカウントで、現在使用しているMicrosoft Entra アカウントに Storage Blob Data Contributor ロールを割り当てます。 アクセス ロールの付与の詳細については、「Azure ポータルを使用してAzureロールを割り当ておよび Microsoft Entra ID を使用して Event Hubs リソースへのアクセス権を認証する」を参照してください。

重要

このチュートリアルの手順を完了するには、Spring Boot 2.5 以上のバージョンが必要です。

ローカル環境を準備する

このチュートリアルでは、コンフィギュレーションとコードには認証操作がありません。 ただし、Azure サービスに接続するには認証が必要です。 認証を完了するには、Azure ID クライアント ライブラリを使用する必要があります。 Spring Cloud Azureでは DefaultAzureCredential を使用します。これは、コードを変更せずに資格情報を取得するのに役立つAzure ID ライブラリが提供します。

DefaultAzureCredential は複数の認証方法をサポートしており、実行時に使用する方法が決定されます。 このアプローチは、環境固有のコードを実装することなく、アプリが異なる環境（ローカル環境や本番環境など）で異なる認証方法を使用することを可能にします。 詳細については、Authenticate Azure ホスト型Java アプリケーションの DefaultAzureCredential セクションを参照してください。

Azure CLI、IntelliJ、またはその他の方法を使用してローカル開発環境で認証を完了するには、「Azure Java開発環境での認証を参照してください。 Azureホスティング環境で認証を完了するには、マネージド ID を使用することをお勧めします。 詳細については、「Azure リソースのマネージド ID とは」を参照してください>

Spring Cloud Azure Event Hubs Starter を使用する

Spring Cloud Azure Event Hubs Starter モジュールは、Spring Boot フレームワークを使用して Event Hubs Java クライアント ライブラリをインポートします。 Spring Cloud AzureとAzure SDKは、相互に排他的でないパターンで一緒に使用できます。 そのため、Spring アプリケーションで Event Hubs Java クライアント API を引き続き使用できます。

依存関係を追加する

Spring Cloud Azure Event Hubs Starter モジュールをインストールするには、次の依存関係を pom.xml ファイルに追加します。

• Spring Cloud Azure BOM（部品表）

  <dependencyManagement>
     <dependencies>
       <dependency>
         <groupId>com.azure.spring</groupId>
         <artifactId>spring-cloud-azure-dependencies</artifactId>
         <version>7.4.0</version>
         <type>pom</type>
         <scope>import</scope>
         </dependency>
     </dependencies>
  </dependencyManagement>
  

  注

  Spring Boot 4.0.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 7.4.0 に設定します。

  Spring Boot 3.5.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 6.5.0 に設定します。

  Spring Boot 3.1.x-3.5.x を使用している場合は、必ず spring-cloud-azure-dependencies バージョンを 5.25.0 に設定してください。

  Spring Boot 2.xを使用している場合は、 spring-cloud-azure-dependencies バージョンを 4.20.0に設定してください。

  この部品表（BOM）は、<dependencyManagement> ファイルの セクションに設定する必要があります。 これにより、すべての Spring Cloud Azure依存関係で同じバージョンが使用されるようになります。

  この BOM に使用されたバージョンの詳細については、「どのバージョンの Spring Cloud Azure を使用すべきか」を参照してください。

• Spring Cloud Azure Event Hubsアーティファクト:

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

メッセージの送受信にアプリケーションをコード化する

このガイドでは、Spring アプリケーションのコンテキストで Event Hubs Java クライアントを使用する方法について説明します。 ここでは、次の2つのオプションを紹介します。

• Spring Boot自動設定を使用し、Springコンテキストからすぐに使用できるクライアントを使用します（推奨）。
• クライアントをプログラム的に構築します。

Spring IoCコンテナからクライアントBeanを自動配線する方法には、次のような利点があります。これにより、Event Hubsクライアントを使用して開発する際に、より柔軟で効率的なエクスペリエンスを提供できます。

• 外部設定を適用して、異なる環境で同じアプリケーションコードを使用できるようにします。
• ビルダパターンを学習し、このクライアントをアプリケーションコンテキストに登録するプロセスをSpring Bootフレームワークに委任できます。 この委任により、自身のビジネス要件を満たすクライアントの使用方法に集中できます。
• ヘルスインジケータを使用して、アプリケーションおよび内部コンポーネントのステータスとヘルスを簡単に検査できます。

次のセクションでは、EventProcessorClientとEventHubProducerClientを2つのオプションとともに使用する方法を示すコード例を示します。

注

Azure Java SDK for Event Hubs には、Event Hubs と対話するための複数のクライアントが用意されています。 また、スターターは、すべてのEvent Hubsクライアントとクライアントビルダに自動設定を提供します。 この記事では、EventProcessorClientとEventHubProducerClientのみを例として使用しています。

Spring Boot Autoconfigurationを使用

Event Hubsとの間でメッセージを送受信するには、次の手順に従ってアプリケーションを設定します。

1. Event HubsのネームスペースとEvent Hubs名を設定するには、次のプロパティ設定を使用します。

   spring.cloud.azure.eventhubs.namespace=<your event-hubs-namespace>
   spring.cloud.azure.eventhubs.event-hub-name=<your-event-hub-name>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=<your-storage-account-name>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=<your-storage-account-container-name>
   spring.cloud.azure.eventhubs.processor.consumer-group=$Default
   

2. 次の例に示すように、新しい EventHubProcessorClientConfiguration Java クラスを作成します。 このクラスは、EventProcessorClientのメッセージとエラーハンドラを登録するために使用されます。

   import com.azure.spring.cloud.service.eventhubs.consumer.EventHubsErrorHandler;
   import com.azure.spring.cloud.service.eventhubs.consumer.EventHubsRecordMessageListener;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.context.annotation.Bean;
   import org.springframework.context.annotation.Configuration;
   
   @Configuration
   public class EventHubProcessorClientConfiguration {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubProcessorClientConfiguration.class);
   
       @Bean
       EventHubsRecordMessageListener processEvent() {
           return eventContext->LOGGER.info("Processing event from partition {} with sequence number {} with body: {}",
               eventContext.getPartitionContext().getPartitionId(), eventContext.getEventData().getSequenceNumber(),
               eventContext.getEventData().getBodyAsString());
       }
   
       @Bean
       EventHubsErrorHandler processError() {
           return errorContext->LOGGER.info("Error occurred in partition processor for partition {}, {}",
               errorContext.getPartitionContext().getPartitionId(),
               errorContext.getThrowable());
       }
   
   }
   

3. 次の例に示すように、SpringアプリケーションにEventProcessorClientとEventHubProducerClientを挿入し、関連するAPIを呼び出してメッセージを送受信します。

   import com.azure.messaging.eventhubs.EventData;
   import com.azure.messaging.eventhubs.EventHubProducerClient;
   import com.azure.messaging.eventhubs.EventProcessorClient;
   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 java.util.Collections;
   import java.util.concurrent.TimeUnit;
   
   @SpringBootApplication
   public class EventHubClientApplication implements CommandLineRunner {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubClientApplication.class);
       private final EventHubProducerClient eventHubProducerClient;
       private final EventProcessorClient eventProcessorClient;
   
       public EventHubClientApplication(EventHubProducerClient eventHubProducerClient,
                                        EventProcessorClient eventProcessorClient) {
           this.eventHubProducerClient = eventHubProducerClient;
           this.eventProcessorClient = eventProcessorClient;
       }
   
       public static void main(String[] args) {
           SpringApplication.run(EventHubClientApplication.class, args);
       }
   
       @Override
       public void run(String... args) throws Exception {
           eventProcessorClient.start();
           // Wait for the processor client to be ready
           TimeUnit.SECONDS.sleep(10);
   
           eventHubProducerClient.send(Collections.singletonList(new EventData("Hello World")));
           LOGGER.info("Successfully sent a message to Event Hubs.");
           eventHubProducerClient.close();
           LOGGER.info("Skip stopping and closing the processor since the processor may not complete the receiving process yet.");
       }
   
   }
   

4. アプリケーションを起動します。 次の例のようなログが表示されます。

   Successfully sent a message to Event Hubs.
   ...
   Processing event from partition 0 with sequence number 0 with body: Hello World
   ...
   Stopping and closing the processor.
   

クライアントをプログラム的に構築します

クライアントBeansは自分で作成できますが、プロセスは複雑です。 Spring Boot アプリケーションでは、プロパティを管理し、ビルダー パターンを学習し、Spring アプリケーション コンテキストにクライアントを登録する必要があります。 次の手順は、その方法を示しています。

1. 次の例に示すように、新しい EventHubClientConfiguration Java クラスを作成します。 このクラスは、 EventProcessorClient と EventHubProducerClient ビーンズを宣言するために使用されます。 <your event-hubs-namespace>、<your-event-hub-name>、<your-storage-account-name>、<your-storage-account-container-name>プレースホルダを必ず実際の値に置き換えてください。

   import com.azure.identity.DefaultAzureCredentialBuilder;
   import com.azure.messaging.eventhubs.EventHubClientBuilder;
   import com.azure.messaging.eventhubs.EventHubProducerClient;
   import com.azure.messaging.eventhubs.EventProcessorClient;
   import com.azure.messaging.eventhubs.EventProcessorClientBuilder;
   import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
   import com.azure.messaging.eventhubs.models.ErrorContext;
   import com.azure.messaging.eventhubs.models.EventContext;
   import com.azure.storage.blob.BlobContainerAsyncClient;
   import com.azure.storage.blob.BlobContainerClientBuilder;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.context.annotation.Bean;
   import org.springframework.context.annotation.Configuration;
   
   @Configuration
   public class EventHubClientConfiguration {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubClientConfiguration.class);
       private static final String EVENT_HUB_FULLY_QUALIFIED_NAMESPACE = "<your event-hubs-namespace>.servicebus.windows.net";
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
       private static final String CONSUMER_GROUP = "$Default";
       private static final String STORAGE_ACCOUNT_ENDPOINT = "https://<your-storage-account-name>.blob.core.windows.net";
       private static final String STORAGE_CONTAINER_NAME = "<your-storage-account-container-name>";
   
       @Bean
       EventHubClientBuilder eventHubClientBuilder() {
           return new EventHubClientBuilder().credential(EVENT_HUB_FULLY_QUALIFIED_NAMESPACE, EVENT_HUB_NAME,
               new DefaultAzureCredentialBuilder()
                   .build());
       }
   
       @Bean
       BlobContainerClientBuilder blobContainerClientBuilder() {
           return new BlobContainerClientBuilder().credential(new DefaultAzureCredentialBuilder()
                                                      .build())
                                                  .endpoint(STORAGE_ACCOUNT_ENDPOINT)
                                                  .containerName(STORAGE_CONTAINER_NAME);
       }
   
       @Bean
       BlobContainerAsyncClient blobContainerAsyncClient(BlobContainerClientBuilder blobContainerClientBuilder) {
           return blobContainerClientBuilder.buildAsyncClient();
       }
   
       @Bean
       EventProcessorClientBuilder eventProcessorClientBuilder(BlobContainerAsyncClient blobContainerAsyncClient) {
           return new EventProcessorClientBuilder().credential(EVENT_HUB_FULLY_QUALIFIED_NAMESPACE, EVENT_HUB_NAME,
                                                       new DefaultAzureCredentialBuilder()
                                                           .build())
                                                   .consumerGroup(CONSUMER_GROUP)
                                                   .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient))
                                                   .processEvent(EventHubClientConfiguration::processEvent)
                                                   .processError(EventHubClientConfiguration::processError);
       }
   
       @Bean
       EventHubProducerClient eventHubProducerClient(EventHubClientBuilder eventHubClientBuilder) {
           return eventHubClientBuilder.buildProducerClient();
   
       }
   
       @Bean
       EventProcessorClient eventProcessorClient(EventProcessorClientBuilder eventProcessorClientBuilder) {
           return eventProcessorClientBuilder.buildEventProcessorClient();
       }
   
       public static void processEvent(EventContext eventContext) {
           LOGGER.info("Processing event from partition {} with sequence number {} with body: {}",
               eventContext.getPartitionContext().getPartitionId(), eventContext.getEventData().getSequenceNumber(),
               eventContext.getEventData().getBodyAsString());
       }
   
       public static void processError(ErrorContext errorContext) {
           LOGGER.info("Error occurred in partition processor for partition {}, {}",
               errorContext.getPartitionContext().getPartitionId(),
               errorContext.getThrowable());
       }
   
   }
   

2. 次の例に示すように、SpringアプリケーションにEventProcessorClientとEventHubProducerClientを挿入します。

   import com.azure.messaging.eventhubs.EventData;
   import com.azure.messaging.eventhubs.EventHubProducerClient;
   import com.azure.messaging.eventhubs.EventProcessorClient;
   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 java.util.Collections;
   import java.util.concurrent.TimeUnit;
   
   @SpringBootApplication
   public class EventHubClientApplication implements CommandLineRunner {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubClientApplication.class);
       private final EventHubProducerClient eventHubProducerClient;
       private final EventProcessorClient eventProcessorClient;
   
       public EventHubClientApplication(EventHubProducerClient eventHubProducerClient,
                                        EventProcessorClient eventProcessorClient) {
           this.eventHubProducerClient = eventHubProducerClient;
           this.eventProcessorClient = eventProcessorClient;
       }
   
       public static void main(String[] args) {
           SpringApplication.run(EventHubClientApplication.class, args);
       }
   
       @Override
       public void run(String... args) throws Exception {
           eventProcessorClient.start();
           // Wait for the processor client to be ready
           TimeUnit.SECONDS.sleep(10);
   
           eventHubProducerClient.send(Collections.singletonList(new EventData("Hello World")));
           LOGGER.info("Successfully sent a message to Event Hubs.");
           eventHubProducerClient.close();
           LOGGER.info("Stopping and closing the processor");
           eventProcessorClient.stop();
       }
   
   }
   

3. アプリケーションを起動します。 次の例のようなログが表示されます。

   Successfully sent a message to Event Hubs.
   ...
   Processing event from partition 0 with sequence number 0 with body: Hello World
   ...
   Stopping and closing the processor.
   

次のリストは、このコードが柔軟でないかグレースフルでない理由を示しています。

• Event HubsのネームスペースとEvent Hubs名はハードコードされています。
• Spring 環境から構成を取得するために @Value を使用する場合、 application.properties ファイルに IDE ヒントを含めることはできません。
• マイクロサービスのシナリオがある場合は、各プロジェクトでコードを複製する必要があります。また、エラーが発生しやすく、一貫性を保つのが困難です。

幸いなことに、Spring Cloud Azureでは、クライアント Bean を自分で構築する必要はありません。 代わりに、これらを直接注入して、すでに慣れている設定プロパティを使用してストレージキューを構成することができます。 詳細については、「Spring Cloud Azure の構成」を参照してください。

Spring Cloud Azureでは、さまざまなシナリオに対して次のグローバル構成も提供されます。 詳細については、「Spring Cloud Azure 構成」の「Azure Service SDK のグローバル構成」セクションを参照してください。

• プロキシ オプション。
• 再試行オプション。
• AMQP トランスポート クライアント オプション。

異なるAzure クラウドに接続することもできます。 詳細については、「異なるAzure クラウドに接続するを参照してください。

Spring MessagingおよびAzure Event Hubsを使用する

Spring Messaging Azure Event Hubs モジュールは、Event Hubs で Spring Messaging フレームワークをサポートします。

Spring Messaging Azure Event Hubsを使用している場合は、次の機能を使用できます。

• EventHubsTemplate：Event Hubsに非同期および同期してメッセージを送信します。
• @EventHubsListener：宛先のEvent Hubsメッセージリスナーのターゲットとなるメソッドをマークします。

このガイドでは、Spring Messaging Azure Event Hubsを使用して Event Hubs との間でメッセージを送受信する方法について説明します。

依存関係を追加する

Spring Messaging Azure Event Hubs モジュールをインストールするには、次の依存関係を pom.xml ファイルに追加します。

• Spring Cloud Azure BOM（部品表）

  <dependencyManagement>
     <dependencies>
       <dependency>
         <groupId>com.azure.spring</groupId>
         <artifactId>spring-cloud-azure-dependencies</artifactId>
         <version>7.4.0</version>
         <type>pom</type>
         <scope>import</scope>
         </dependency>
     </dependencies>
  </dependencyManagement>
  

  注

  Spring Boot 4.0.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 7.4.0 に設定します。

  Spring Boot 3.5.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 6.5.0 に設定します。

  Spring Boot 3.1.x-3.5.x を使用している場合は、必ず spring-cloud-azure-dependencies バージョンを 5.25.0 に設定してください。

  Spring Boot 2.xを使用している場合は、 spring-cloud-azure-dependencies バージョンを 4.20.0に設定してください。

  この部品表（BOM）は、<dependencyManagement> ファイルの セクションに設定する必要があります。 これにより、すべての Spring Cloud Azure依存関係で同じバージョンが使用されるようになります。

  この BOM に使用されたバージョンの詳細については、「どのバージョンの Spring Cloud Azure を使用すべきか」を参照してください。

• Spring Cloud Azure スターター、Spring Messaging Event Hubs、Azure Event Hubs チェックポイント ストアのアーティファクト:

  <dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter</artifactId>
  </dependency>
  <dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-messaging-azure-eventhubs</artifactId>
  </dependency>
  <dependency>
    <groupId>com.azure</groupId>
    <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
  </dependency>
  

メッセージの送受信にアプリケーションをコード化する

Event Hubsとの間でメッセージを送受信するには、次の手順に従ってアプリケーションを設定します。

1. 次のプロパティ設定を使用して、Event Hubsネームスペースとストレージブロブを構成します。

   spring.cloud.azure.eventhubs.namespace=<your event-hubs-namespace>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=<your-storage-account-name>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=<your-storage-account-container-name>
   

2. 次の例に示すように、新しい ConsumerService Java クラスを作成します。 このクラスは、メッセージ受信者を定義するために使用されます。 <your-event-hub-name>プレースホルダを実際の値に置き換えてください。

   import com.azure.spring.messaging.eventhubs.implementation.core.annotation.EventHubsListener;
   import org.springframework.stereotype.Service;
   
   @Service
   public class ConsumerService {
   
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
       private static final String CONSUMER_GROUP = "$DEFAULT";
   
       @EventHubsListener(destination = EVENT_HUB_NAME, group = CONSUMER_GROUP)
       public void handleMessageFromEventHub(String message) {
           System.out.printf("New message received: %s%n", message);
       }
   
   }
   

3. 次の例に示すように、Springでメッセージを送受信するために、送信者と受信者を有線で接続します。 <your-event-hub-name>プレースホルダを実際の値に置き換えてください。

   import com.azure.spring.messaging.eventhubs.core.EventHubsTemplate;
   import com.azure.spring.messaging.implementation.annotation.EnableAzureMessaging;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.boot.SpringApplication;
   import org.springframework.boot.autoconfigure.SpringBootApplication;
   import org.springframework.context.ConfigurableApplicationContext;
   import org.springframework.messaging.support.MessageBuilder;
   
   @SpringBootApplication
   @EnableAzureMessaging
   public class EventHubMessagingApplication {
   
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubMessagingApplication.class);
   
       public static void main(String[] args) {
           ConfigurableApplicationContext applicationContext = SpringApplication.run(EventHubMessagingApplication.class);
           EventHubsTemplate eventHubsTemplate = applicationContext.getBean(EventHubsTemplate.class);
           LOGGER.info("Sending a message to the Event Hubs.");
           eventHubsTemplate.sendAsync(EVENT_HUB_NAME, MessageBuilder.withPayload("Hello world").build()).subscribe();
       }
   
   }
   

   ヒント

   @EnableAzureMessaging注釈を追加することを忘れないでください。これは@EventHubsListenerで注釈されたメソッドの発見をトリガーし、メッセージリスナーコンテナをカバー下に作成します。

4. アプリケーションを起動します。 次の例のようなログが表示されます。

   Sending a message to the Event Hubs.
   New message received: Hello world
   

Spring Integration Azure Event Hubsを使用する

Spring Integration Azure Event Hubs モジュールは、Event Hubs を使用した Spring Integration フレームワークのサポートを提供します。

SpringアプリケーションがSpring Integrationメッセージチャネルを使用している場合は、チャネルアダプタを使用してメッセージチャネルとEvent Hubs間でメッセージをルーティングできます。

インバウンドチャネルアダプタは、Event Hubsからメッセージチャネルにメッセージを転送します。 アウトバウンドチャネルアダプタは、メッセージチャネルからEvent Hubsにメッセージをパブリッシュします。

このガイドでは、Spring Integration Azure Event Hubsを使用して Event Hubs との間でメッセージを送受信する方法について説明します。

依存関係を追加する

Spring Cloud Azure Event Hubs Integration Starter モジュールをインストールするには、pom.xml ファイルに次の依存関係を追加します。

• Spring Cloud Azure BOM（部品表）

  <dependencyManagement>
     <dependencies>
       <dependency>
         <groupId>com.azure.spring</groupId>
         <artifactId>spring-cloud-azure-dependencies</artifactId>
         <version>7.4.0</version>
         <type>pom</type>
         <scope>import</scope>
         </dependency>
     </dependencies>
  </dependencyManagement>
  

  注

  Spring Boot 4.0.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 7.4.0 に設定します。

  Spring Boot 3.5.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 6.5.0 に設定します。

  Spring Boot 3.1.x-3.5.x を使用している場合は、必ず spring-cloud-azure-dependencies バージョンを 5.25.0 に設定してください。

  Spring Boot 2.xを使用している場合は、 spring-cloud-azure-dependencies バージョンを 4.20.0に設定してください。

  この部品表（BOM）は、<dependencyManagement> ファイルの セクションに設定する必要があります。 これにより、すべての Spring Cloud Azure依存関係で同じバージョンが使用されるようになります。

  この BOM に使用されたバージョンの詳細については、「どのバージョンの Spring Cloud Azure を使用すべきか」を参照してください。

• Spring Cloud Azure Event Hubs 統合アーティファクト:

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

メッセージの送受信にアプリケーションをコード化する

Event Hubsとの間でメッセージを送受信するには、次の手順に従ってアプリケーションを設定します。

1. 次のプロパティ設定を使用して、Event Hubsネームスペースとストレージブロブを構成します。

   spring.cloud.azure.eventhubs.namespace=<your event-hubs-namespace>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=<your-storage-account-name>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=<your-storage-account-container-name>
   

2. 次の例に示すように、新しい MessageReceiveConfiguration Java クラスを作成します。 このクラスは、メッセージ受信者を定義するために使用されます。 <your-event-hub-name>プレースホルダを実際の値に置き換えてください。

   import com.azure.spring.integration.eventhubs.inbound.EventHubsInboundChannelAdapter;
   import com.azure.spring.messaging.eventhubs.core.EventHubsProcessorFactory;
   import com.azure.spring.messaging.eventhubs.core.checkpoint.CheckpointConfig;
   import com.azure.spring.messaging.eventhubs.core.checkpoint.CheckpointMode;
   import com.azure.spring.messaging.eventhubs.core.listener.EventHubsMessageListenerContainer;
   import com.azure.spring.messaging.eventhubs.core.properties.EventHubsContainerProperties;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.beans.factory.annotation.Qualifier;
   import org.springframework.context.annotation.Bean;
   import org.springframework.context.annotation.Configuration;
   import org.springframework.integration.annotation.ServiceActivator;
   import org.springframework.integration.channel.DirectChannel;
   import org.springframework.messaging.MessageChannel;
   
   @Configuration
   public class MessageReceiveConfiguration {
   
       private static final String INPUT_CHANNEL = "input";
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
       private static final String CONSUMER_GROUP = "$Default";
       private static final Logger LOGGER = LoggerFactory.getLogger(MessageReceiveConfiguration.class);
   
       @ServiceActivator(inputChannel = INPUT_CHANNEL)
       public void messageReceiver(byte[] payload) {
           String message = new String(payload);
           LOGGER.info("New message received: {}", message);
       }
   
       @Bean
       public EventHubsMessageListenerContainer messageListenerContainer(EventHubsProcessorFactory processorFactory) {
           EventHubsContainerProperties containerProperties = new EventHubsContainerProperties();
           containerProperties.setEventHubName(EVENT_HUB_NAME);
           containerProperties.setConsumerGroup(CONSUMER_GROUP);
           containerProperties.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));
           return new EventHubsMessageListenerContainer(processorFactory, containerProperties);
       }
   
       @Bean
       public EventHubsInboundChannelAdapter messageChannelAdapter(@Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
                                                                   EventHubsMessageListenerContainer listenerContainer) {
           EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(listenerContainer);
           adapter.setOutputChannel(inputChannel);
           return adapter;
       }
   
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   
   }
   

3. 次の例に示すように、新しい MessageSendConfiguration Java クラスを作成します。 このクラスは、メッセージ送信者を定義するために使用されます。 <your-event-hub-name>プレースホルダを実際の値に置き換えてください。

   import com.azure.spring.integration.core.handler.DefaultMessageHandler;
   import com.azure.spring.messaging.eventhubs.core.EventHubsTemplate;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import org.springframework.context.annotation.Bean;
   import org.springframework.context.annotation.Configuration;
   import org.springframework.integration.annotation.MessagingGateway;
   import org.springframework.integration.annotation.ServiceActivator;
   import org.springframework.messaging.MessageHandler;
   import org.springframework.util.concurrent.ListenableFutureCallback;
   
   @Configuration
   public class MessageSendConfiguration {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(MessageSendConfiguration.class);
       private static final String OUTPUT_CHANNEL = "output";
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
   
       @Bean
       @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
       public MessageHandler messageSender(EventHubsTemplate eventHubsTemplate) {
           DefaultMessageHandler handler = new DefaultMessageHandler(EVENT_HUB_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;
       }
   
       @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
       public interface EventHubOutboundGateway {
           void send(String text);
       }
   
   }
   

4. 次の例に示すように、Springでメッセージを送受信するために、送信者と受信者を有線で接続します。

   import org.springframework.boot.SpringApplication;
   import org.springframework.boot.autoconfigure.SpringBootApplication;
   import org.springframework.context.ConfigurableApplicationContext;
   import org.springframework.context.annotation.Configuration;
   import org.springframework.integration.config.EnableIntegration;
   
   @SpringBootApplication
   @EnableIntegration
   @Configuration(proxyBeanMethods = false)
   public class EventHubIntegrationApplication {
   
       public static void main(String[] args) {
           ConfigurableApplicationContext applicationContext = SpringApplication.run(EventHubIntegrationApplication.class, args);
           MessageSendConfiguration.EventHubOutboundGateway outboundGateway = applicationContext.getBean(MessageSendConfiguration.EventHubOutboundGateway.class);
           outboundGateway.send("Hello World");
       }
   }
   

   ヒント

   Spring Integrationインフラストラクチャを有効にする@EnableIntegration注釈を追加することを忘れないでください。

5. アプリケーションを起動します。 次の例のようなログが表示されます。

   Message was sent successfully.
   New message received: Hello World
   

Spring Cloud Azure Stream Event Hubs Binder を使用する

Spring Cloud Stream アプリケーションで Event Hubs API を呼び出すには、Spring Cloud Azure Event Hubs Stream Binder モジュールを使用します。

この取扱説明書では、Spring Cloud Stream Event Hubs Binderを使用して、Event Hubsとの間でメッセージを送受信する方法について説明します。

依存関係を追加する

Spring Cloud Azure Event Hubs Stream Binder モジュールをインストールするには、次の依存関係を pom.xml ファイルに追加します。

• Spring Cloud Azure BOM（部品表）

  <dependencyManagement>
     <dependencies>
       <dependency>
         <groupId>com.azure.spring</groupId>
         <artifactId>spring-cloud-azure-dependencies</artifactId>
         <version>7.4.0</version>
         <type>pom</type>
         <scope>import</scope>
         </dependency>
     </dependencies>
  </dependencyManagement>
  

  注

  Spring Boot 4.0.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 7.4.0 に設定します。

  Spring Boot 3.5.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 6.5.0 に設定します。

  Spring Boot 3.1.x-3.5.x を使用している場合は、必ず spring-cloud-azure-dependencies バージョンを 5.25.0 に設定してください。

  Spring Boot 2.xを使用している場合は、 spring-cloud-azure-dependencies バージョンを 4.20.0に設定してください。

  この部品表（BOM）は、<dependencyManagement> ファイルの セクションに設定する必要があります。 これにより、すべての Spring Cloud Azure依存関係で同じバージョンが使用されるようになります。

  この BOM に使用されたバージョンの詳細については、「どのバージョンの Spring Cloud Azure を使用すべきか」を参照してください。

• Spring Cloud Azure Event Hubs Stream Binder アーティファクト:

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

メッセージの送受信にアプリケーションをコード化する

Event Hubsとの間でメッセージを送受信するには、次の手順に従ってアプリケーションを設定します。

1. 次のプロパティ設定を使用して、Event Hubsネームスペースとストレージブロブを構成します。

   spring.cloud.azure.eventhubs.namespace=<your event-hubs-namespace>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name=<your-storage-account-name>
   spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name=<your-storage-account-container-name>
   

2. メッセージ受信者を作成します。

   アプリケーションをイベントシンクとして使用するには、次のタスクを実行して入力バインダを設定します。

   • メッセージ処理ロジックを定義する Consumer Beanを宣言します。 たとえば、次の Consumer Beanは consumeという名前です。

     @Bean
     public Consumer<Message<String>> consume() {
          return message -> {
              System.out.printf("New message received: %s%n", message.getPayload());
          };
     }
     

   • 次の構成を追加して、使用する Event Hub 名を指定します。 <your-event-hub-name>プレースホルダを実際の値に置き換えてください。

     # name for the above `Consumer` bean
     spring.cloud.stream.function.definition=consume
     spring.cloud.stream.bindings.consume-in-0.destination=<your-event-hub-name>
     spring.cloud.stream.bindings.consume-in-0.group=$Default
     spring.cloud.stream.eventhubs.bindings.consume-in-0.consumer.checkpoint.mode=MANUAL
     

3. メッセージ送信者を作成します。

   アプリケーションをイベントソースとして使用するには、次のタスクを実行して出力バインダを設定します。

   • 次の例に示すように、アプリケーション内でメッセージの送信元を定義するSupplierBeanを定義します。

     @Bean
     public Supplier<Message<String>> supply() {
         return () -> {
             System.out.println("Sending a message.");
             return MessageBuilder.withPayload("Hello world").build();
         };
     }
     

   • 次のコンフィギュレーションを追加して、送信するEvent Hub名前を指定します。 <your-event-hub-name>プレースホルダを実際の値に置き換えてください。

     # "consume" is added from the above step
     spring.cloud.stream.function.definition=consume;supply
     spring.cloud.stream.bindings.supply-out-0.destination=<your-event-hub-name>
     

4. アプリケーションを起動します。 次の例のようなログが表示されます。

   Sending a message.
   New message received: Hello world.
   

Azure Event Hubsで Spring Kafka を使用する

Event Hubsは、既存のKafkaベースのアプリケーションが使用できるKafkaエンドポイントを提供します。 このアプローチは、独自のKafkaクラスタを実行する代わりになります。 Event Hubs は、既存の Kafka アプリケーションの多くで動作します。 詳細については、Apache Kafka 用の Event Hubs に関するページを参照してください。

このガイドでは、Azure Event Hubs と Spring Kafka を使用して、Event Hubs との間でメッセージを送受信する方法について説明します。

依存関係を追加する

Spring Cloud Azure starter モジュールと Spring Kafka モジュールをインストールするには、次の依存関係を pom.xml ファイルに追加します。

• Spring Cloud Azure BOM（部品表）

  <dependencyManagement>
     <dependencies>
       <dependency>
         <groupId>com.azure.spring</groupId>
         <artifactId>spring-cloud-azure-dependencies</artifactId>
         <version>7.4.0</version>
         <type>pom</type>
         <scope>import</scope>
         </dependency>
     </dependencies>
  </dependencyManagement>
  

  注

  Spring Boot 4.0.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 7.4.0 に設定します。

  Spring Boot 3.5.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 6.5.0 に設定します。

  Spring Boot 3.1.x-3.5.x を使用している場合は、必ず spring-cloud-azure-dependencies バージョンを 5.25.0 に設定してください。

  Spring Boot 2.xを使用している場合は、 spring-cloud-azure-dependencies バージョンを 4.20.0に設定してください。

  この部品表（BOM）は、<dependencyManagement> ファイルの セクションに設定する必要があります。 これにより、すべての Spring Cloud Azure依存関係で同じバージョンが使用されるようになります。

  この BOM に使用されたバージョンの詳細については、「どのバージョンの Spring Cloud Azure を使用すべきか」を参照してください。

• Spring Cloud Azure スターターと Spring Kafka アーティファクト:

  <dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
  </dependency>
  

メッセージの送受信にアプリケーションをコード化する

Event Hubsとの間でメッセージを送受信するには、次の手順に従ってアプリケーションを設定します。

1. Event Hubsのネームスペースを設定するには、次のプロパティ設定を使用します。

   spring.kafka.bootstrap-servers=<your event-hubs-namespace>.servicebus.windows.net:9093
   

2. 次の例のように、KafkaTemplateを使ってメッセージを送信し、@KafkaListenerを使ってメッセージを受信します。 <your-event-hub-name>プレースホルダを実際の値に置き換えてください。

   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.kafka.annotation.KafkaListener;
   import org.springframework.kafka.core.KafkaTemplate;
   
   @SpringBootApplication
   public class EventHubKafkaApplication implements CommandLineRunner {
   
       private static final Logger LOGGER = LoggerFactory.getLogger(EventHubKafkaApplication.class);
       private static final String EVENT_HUB_NAME = "<your-event-hub-name>";
       private static final String CONSUMER_GROUP = "$Default";
       private final KafkaTemplate<String, String> kafkaTemplate;
   
       public EventHubKafkaApplication(KafkaTemplate<String, String> kafkaTemplate) {
           this.kafkaTemplate = kafkaTemplate;
       }
   
       public static void main(String[] args) {
           SpringApplication.run(EventHubKafkaApplication.class, args);
       }
   
       @Override
       public void run(String... args) {
           kafkaTemplate.send(EVENT_HUB_NAME, "Hello World");
           LOGGER.info("Message was sent successfully.");
       }
   
       @KafkaListener(topics = EVENT_HUB_NAME, groupId = CONSUMER_GROUP)
       public void receive(String message) {
           LOGGER.info("New message received: {}", message);
       }
   
   }
   

3. アプリケーションを起動します。 次の例のようなログが表示されます。

   Message was sent successfully.
   New message received: Hello world
   

Azure Event Hubsで Spring Cloud Stream Kafka Binder を使用する

Spring Cloud Streamは、アプリケーション開発者がメッセージ主導のマイクロサービスを作成できるようにするフレームワークです。 メッセージングシステムとSpring Cloud Streamの間のブリッジは、バインダ抽象化によって行われます。 バインダはいくつかのメッセージングシステム用に存在しますが、最も一般的に使用されるバインダの1つはApache Kafkaです。

このガイドでは、Azure Event Hubsと Spring Cloud Stream Kafka Binder を使用して、Event Hubs との間でメッセージを送受信する方法について説明します。

依存関係を追加する

Spring Cloud Azure starter および Spring Cloud Stream バインダー Kafka モジュールをインストールするには、次の依存関係を pom.xml ファイルに追加します。

• Spring Cloud Azure BOM（部品表）

  <dependencyManagement>
     <dependencies>
       <dependency>
         <groupId>com.azure.spring</groupId>
         <artifactId>spring-cloud-azure-dependencies</artifactId>
         <version>7.4.0</version>
         <type>pom</type>
         <scope>import</scope>
         </dependency>
     </dependencies>
  </dependencyManagement>
  

  注

  Spring Boot 4.0.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 7.4.0 に設定します。

  Spring Boot 3.5.x を使用している場合は、 spring-cloud-azure-dependencies バージョンを 6.5.0 に設定します。

  Spring Boot 3.1.x-3.5.x を使用している場合は、必ず spring-cloud-azure-dependencies バージョンを 5.25.0 に設定してください。

  Spring Boot 2.xを使用している場合は、 spring-cloud-azure-dependencies バージョンを 4.20.0に設定してください。

  この部品表（BOM）は、<dependencyManagement> ファイルの セクションに設定する必要があります。 これにより、すべての Spring Cloud Azure依存関係で同じバージョンが使用されるようになります。

  この BOM に使用されたバージョンの詳細については、「どのバージョンの Spring Cloud Azure を使用すべきか」を参照してください。

• Spring Cloud Azure スターターアーティファクト:

  <dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
  </dependency>
  

メッセージの送受信にアプリケーションをコード化する

Event Hubsとの間でメッセージを送受信するには、次の手順に従ってアプリケーションを設定します。

1. 次のプロパティ設定を使用して、Kafkaブローカを構成します。

   spring.cloud.stream.kafka.binder.brokers=<your event-hubs-namespace>.servicebus.windows.net:9093
   

2. メッセージ受信者を作成します。

   アプリケーションをイベントシンクとして使用するには、次のタスクを実行して入力バインダを設定します。

   • メッセージ処理ロジックを定義する Consumer Beanを宣言します。 たとえば、次の Consumer Beanは consumeという名前です。

     @Bean
     public Consumer<Message<String>> consume() {
         return message -> {
             System.out.printf("New message received: %s%n", message.getPayload());
         };
     }
     

   • 次の構成を追加して、使用する Event Hub 名を指定します。 <your-event-hub-name>プレースホルダを実際の値に置き換えてください。

     # name for the above `Consumer` bean
     spring.cloud.stream.function.definition=consume
     spring.cloud.stream.bindings.consume-in-0.destination=<your-event-hub-name>
     spring.cloud.stream.bindings.consume-in-0.group=$Default
     

3. メッセージ送信者を作成します。

   アプリケーションをイベントソースとして使用するには、次のタスクを実行して出力バインダを設定します。

   • 次の例に示すように、アプリケーション内でメッセージの送信元を定義するSupplierBeanを定義します。

     @Bean
     public Supplier<Message<String>> supply() {
         return () -> {
             System.out.println("Sending a message.");
             return MessageBuilder.withPayload("Hello world").build();
         };
     }
     

   • 次のコンフィギュレーションを追加して、送信するEvent Hub名前を指定します。 <your-event-hub-name>プレースホルダを実際の値に置き換えてください。

     # "consume" is added from the above step
     spring.cloud.stream.function.definition=consume;supply
     spring.cloud.stream.bindings.supply-out-0.destination=<your-event-hub-name>
     

4. アプリケーションを起動します。 次の例のようなログが表示されます。

   Sending a message.
   New message received: Hello world.
   

Azure Spring Appsにデプロイする

Spring Boot アプリケーションがローカルで実行されたので、運用環境に移行します。 Azure Spring Apps を使用すると、コードを変更することなく、Spring Boot アプリケーションをAzureに簡単にデプロイできます。 Spring アプリケーションのインフラストラクチャはこのサービスによって管理されるため、開発者はコードに専念できます。 Azure Spring Appsでは、包括的な監視と診断、構成管理、サービス検出、CI/CD 統合、ブルーグリーンデプロイなどを使用したライフサイクル管理が提供されます。 アプリケーションをAzure Spring Appsにデプロイするには、「初めてのアプリケーションを Azure Spring Apps にデプロイする」を参照してください。

次のステップ

Spring 開発者向けのAzureSpring Cloud Azure Event Hubs サンプル

関連項目

Microsoft Azureで使用できる他の Spring Boot Starter の詳細については、「 Spring Cloud Azureとは