Java を使用して Azure Event Hubs との間でイベントを送受信する

		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs</artifactId>
		    <version>5.18.0</version>
		</dependency>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-identity</artifactId>
		    <version>1.11.2</version>
		    <scope>compile</scope>
		</dependency>

		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs</artifactId>
		    <version>5.18.0</version>
		</dependency>

Microsoft Entra ユーザーにロールを割り当てる

ローカルでの開発時には、Azure Event Hubs に接続するユーザー アカウントに正しいアクセス許可があることを確認してください。 メッセージを送受信するには、Azure Event Hubs データ所有者ロールが必要です。 このロールを自分に割り当てるには、ユーザー アクセス管理者ロール、または Microsoft.Authorization/roleAssignments/write アクションを含む別のロールが必要です。 Azure portal、Azure CLI、または Azure PowerShell を使用して、ユーザーに Azure RBAC ロールを割り当てることができます。 ロールの割り当てに使用できるスコープの詳細は、スコープの概要ページを参照してください。

次の例では、ユーザー アカウントに Azure Event Hubs Data Owner ロールを割り当てます。これにより、Azure Event Hubs リソースにフル アクセスできます。 実際のシナリオでは、より安全な運用環境を実現するため、最小限の特権の原則に従って、必要な最小限のアクセス許可のみをユーザーに付与します。

Azure Event Hubs の Azure の組み込みロール

Azure Event Hubs の場合、Azure portal および Azure リソース管理 API による名前空間とそれに関連するすべてのリソースの管理は、Azure RBAC モデルを使用して既に保護されています。 Azure では、Event Hubs 名前空間へのアクセス権を付与するために、次の Azure 組み込みロールが提供されています。

• Azure Event Hubs データ所有者: Event Hubs 名前空間とそのエンティティ (キュー、トピック、サブスクリプション、フィルター) へのデータ アクセスが可能です。
• Azure Event Hubs データ送信者: このロールを使用して、Event Hubs 名前空間とそのエンティティへのアクセス権を送信者に付与します。
• Azure Event Hubs データ受信者: このロールを使用して、Event Hubs 名前空間とそのエンティティへのアクセス権を受信者に付与します。

カスタム ロールを作成する場合は、Event Hubs 操作に必要な権限に関するページを参照してください。

az eventhubs namespace show -g '<your-event-hub-resource-group>' -n '<your-event-hub-name> --query id

az role assignment create --assignee "<user@domain>" \
--role "Azure Event Hubs Data Owner" \
--scope "<your-resource-id>"

Get-AzResource -ResourceGroupName "<your-event-hub-resource-group>" -Name "<your-event-hub-name>"

New-AzRoleAssignment -SignInName <user@domain> `
-RoleDefinitionName "Azure Event Hubs Data Owner" `
-Scope <yourResourceId>

接続文字列を取得する

新しい名前空間を作成すると、主キーとセカンダリ キー、およびそれぞれが名前空間のすべての側面を完全に制御できる接続文字列を使用して、Shared Access Signature (SAS) の初期ポリシーが自動的に生成されます。 通常の送信者と受信者を対象とした、より権限を制限した規則を作成する方法については、Event Hubs の認証と承認に関するページを参照してください。

クライアントは、接続文字列を使用して Event Hubs の名前空間に接続できます。 名前空間のプライマリ接続文字列をコピーするには、次の手順に従います。

1. [Event Hub 名前空間] ページで、左側のメニューの [共有アクセス ポリシー] を選択します。

2. [共有アクセス ポリシー] ページで、 [RootManageSharedAccessKey] を選択します。

3. [ポリシー: RootManageSharedAccessKey] ウィンドウで、[プライマリ接続文字列] の横にある [コピー] ボタンを選択し、後で使用するために接続文字列をクリップボードにコピーします。 この値をメモ帳などに一時的に貼り付けます。

   

   このページを使用して、主キー、2 次キー、プライマリ接続文字列、セカンダリ接続文字列をコピーできます。

Sender という名前のクラスを追加し、このクラスに次のコードを追加します。

package ehubquickstart;

import com.azure.messaging.eventhubs.*;
import java.util.Arrays;
import java.util.List;

import com.azure.identity.*;

public class SenderAAD {

    // replace <NAMESPACE NAME> with the name of your Event Hubs namespace.
    // Example: private static final String namespaceName = "contosons.servicebus.windows.net";
    private static final String namespaceName = "<NAMESPACE NAME>.servicebus.windows.net";

    // Replace <EVENT HUB NAME> with the name of your event hub. 
    // Example: private static final String eventHubName = "ordersehub";
    private static final String eventHubName = "<EVENT HUB NAME>";

    public static void main(String[] args) {
        publishEvents();
    }
    /**
     * Code sample for publishing events.
     * @throws IllegalArgumentException if the EventData is bigger than the max batch size.
     */
    public static void publishEvents() {
        // create a token using the default Azure credential        
        DefaultAzureCredential credential = new DefaultAzureCredentialBuilder()
                .authorityHost(AzureAuthorityHosts.AZURE_PUBLIC_CLOUD)
                .build();

        // create a producer client        
        EventHubProducerClient producer = new EventHubClientBuilder()        
            .fullyQualifiedNamespace(namespaceName)
            .eventHubName(eventHubName)
            .credential(credential)
            .buildProducerClient();

        // sample events in an array
        List<EventData> allEvents = Arrays.asList(new EventData("Foo"), new EventData("Bar"));

        // create a batch
        EventDataBatch eventDataBatch = producer.createBatch();

        for (EventData eventData : allEvents) {
            // try to add the event from the array to the batch
            if (!eventDataBatch.tryAdd(eventData)) {
                // if the batch is full, send it and then create a new batch
                producer.send(eventDataBatch);
                eventDataBatch = producer.createBatch();

                // Try to add that event that couldn't fit before.
                if (!eventDataBatch.tryAdd(eventData)) {
                    throw new IllegalArgumentException("Event is too large for an empty batch. Max size: "
                        + eventDataBatch.getMaxSizeInBytes());
                }
            }
        }
        // send the last batch of remaining events
        if (eventDataBatch.getCount() > 0) {
            producer.send(eventDataBatch);
        }
        producer.close();
    }   
}

Sender という名前のクラスを追加し、このクラスに次のコードを追加します。

import com.azure.messaging.eventhubs.*;
import java.util.Arrays;
import java.util.List;

public class Sender {
    private static final String connectionString = "<Event Hubs namespace connection string>";
    private static final String eventHubName = "<Event hub name>";

    public static void main(String[] args) {
        publishEvents();
    }
}

イベントをイベント ハブに発行するためのコードを追加する

publishEvents という名前のメソッドを Sender クラスに追加します。

    /**
     * Code sample for publishing events.
     * @throws IllegalArgumentException if the EventData is bigger than the max batch size.
     */
    public static void publishEvents() {
        // create a producer client
        EventHubProducerClient producer = new EventHubClientBuilder()
            .connectionString(connectionString, eventHubName)
            .buildProducerClient();

        // sample events in an array
        List<EventData> allEvents = Arrays.asList(new EventData("Foo"), new EventData("Bar"));

        // create a batch
        EventDataBatch eventDataBatch = producer.createBatch();

        for (EventData eventData : allEvents) {
            // try to add the event from the array to the batch
            if (!eventDataBatch.tryAdd(eventData)) {
                // if the batch is full, send it and then create a new batch
                producer.send(eventDataBatch);
                eventDataBatch = producer.createBatch();

                // Try to add that event that couldn't fit before.
                if (!eventDataBatch.tryAdd(eventData)) {
                    throw new IllegalArgumentException("Event is too large for an empty batch. Max size: "
                        + eventDataBatch.getMaxSizeInBytes());
                }
            }
        }
        // send the last batch of remaining events
        if (eventDataBatch.getCount() > 0) {
            producer.send(eventDataBatch);
        }
        producer.close();
    }

ローカルで開発する場合は、BLOB データにアクセスするユーザー アカウントに正しいアクセス許可があることを確認します。 BLOB データの読み取りと書き込みを行うには、ストレージ BLOB データ共同作成者が必要です。 このロールを自分に割り当てるには、ユーザー アクセス管理者ロール、または Microsoft.Authorization/roleAssignments/write アクションを含む別のロールに割り当てられている必要があります。 Azure portal、Azure CLI、または Azure PowerShell を使用して、ユーザーに Azure RBAC ロールを割り当てることができます。 ロールの割り当てに使用できるスコープの詳細は、スコープの概要ページでご覧いただけます。

このシナリオでは、最小限の特権の原則に従って、ストレージ アカウントに限定したアクセス許可をユーザー アカウントに割り当てます。 この方法を使って、ユーザーに必要最小限のアクセス許可のみを与え、より安全な運用環境を作成します。

次の例では、ストレージ BLOB データ共同作成者ロールを自分のユーザー アカウントに割り当てます。これにより、そのストレージ アカウント内の BLOB データに対する読み取りと書き込みの両方のアクセス権が付与されます。

az storage account show --resource-group '<your-resource-group-name>' --name '<your-storage-account-name>' --query id

az role assignment create --assignee "<user@domain>" \
    --role "Storage Blob Data Contributor" \
    --scope "<your-resource-id>"

Get-AzResource -ResourceGroupName "<yourResourceGroupname>" -Name "<yourStorageAccountName>"

New-AzRoleAssignment -SignInName <user@domain> `
    -RoleDefinitionName "Storage Blob Data Contributor" `
    -Scope <yourStorageAccountId>

ストレージ アカウントへの接続文字列を取得します。

接続文字列とコンテナー名を書き留めておきます。 これらは、受信コードで使用します。

pom.xml ファイルに次の依存関係を追加します。

• azure-messaging-eventhubs
• azure-messaging-eventhubs-checkpointstore-blob
• azure-identity

	<dependencies>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs</artifactId>
		    <version>5.15.0</version>
		</dependency>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
		    <version>1.16.1</version>
		</dependency>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-identity</artifactId>
		    <version>1.8.0</version>
		    <scope>compile</scope>
		</dependency>	
	</dependencies>

pom.xml ファイルに次の依存関係を追加します。

• azure-messaging-eventhubs
• azure-messaging-eventhubs-checkpointstore-blob

	<dependencies>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs</artifactId>
		    <version>5.15.0</version>
		</dependency>
		<dependency>
		    <groupId>com.azure</groupId>
		    <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
		    <version>1.16.1</version>
		</dependency>
	</dependencies>

1. 次の import ステートメントを Java ファイルの先頭に追加します。

   import com.azure.messaging.eventhubs.*;
   import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
   import com.azure.messaging.eventhubs.models.*;
   import com.azure.storage.blob.*;
   import java.util.function.Consumer;
   
   import com.azure.identity.*;
   

2. Receiver という名前のクラスを作成し、そのクラスに次の文字列変数を追加します。 プレースホルダーを適切な値に置き換えます。

   重要

   プレースホルダーを適切な値に置き換えます。

   • <NAMESPACE NAME> をご利用の Event Hubs 名前空間の名前に置換します。
   • <EVENT HUB NAME> を、名前空間内でご利用のイベント ハブの名前で置き換えます。

   private static final String namespaceName = "<NAMESPACE NAME>.servicebus.windows.net";
   private static final String eventHubName = "<EVENT HUB NAME>";
   

3. クラスに次の main メソッドを追加します。

   重要

   プレースホルダーを適切な値に置き換えます。

   • <STORAGE ACCOUNT NAME> は、実際の Azure Storage アカウントの名前にします。
   • <CONTAINER NAME> を、ストレージ アカウントの BLOB コンテナーの名前にします。

   // create a token using the default Azure credential
   DefaultAzureCredential credential = new DefaultAzureCredentialBuilder()
           .authorityHost(AzureAuthorityHosts.AZURE_PUBLIC_CLOUD)
           .build();
   
   // Create a blob container client that you use later to build an event processor client to receive and process events
   BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder()
           .credential(credential)
           .endpoint("https://<STORAGE ACCOUNT NAME>.blob.core.windows.net")
           .containerName("<CONTAINER NAME>")
           .buildAsyncClient();
   
   // Create an event processor client to receive and process events and errors.
   EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder()
       .fullyQualifiedNamespace(namespaceName)
       .eventHubName(eventHubName)
       .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
       .processEvent(PARTITION_PROCESSOR)
       .processError(ERROR_HANDLER)
       .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient))            
       .credential(credential)
       .buildEventProcessorClient();
   
   System.out.println("Starting event processor");
   eventProcessorClient.start();
   
   System.out.println("Press enter to stop.");
   System.in.read();
   
   System.out.println("Stopping event processor");
   eventProcessorClient.stop();
   System.out.println("Event processor stopped.");
   
   System.out.println("Exiting process");  
   

1. 次の import ステートメントを Java ファイルの先頭に追加します。

   import com.azure.messaging.eventhubs.*;
   import com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore;
   import com.azure.messaging.eventhubs.models.*;
   import com.azure.storage.blob.*;
   import java.util.function.Consumer;
   

2. Receiver という名前のクラスを作成し、そのクラスに次の文字列変数を追加します。 プレースホルダーを適切な値に置き換えます。

   重要

   プレースホルダーを適切な値に置き換えます。

   • <Event Hubs namespace connection string> を、Event Hubs 名前空間への接続文字列で置き換えます。 更新
   • <Event hub name> を、名前空間内でご利用のイベント ハブの名前で置き換えます。
   • <Storage connection string> を、Azure ストレージ アカウントへの接続文字列で置き換えます。
   • <Storage container name> を、Azure Blob Storage 内のコンテナーの名前で置き換えます。

   private static final String connectionString = "<Event Hubs namespace connection string>";
   private static final String eventHubName = "<Event hub name>";
   private static final String storageConnectionString = "<Storage connection string>";
   private static final String storageContainerName = "<Storage container name>";
   

3. クラスに次の main メソッドを追加します。

   public static void main(String[] args) throws Exception {
       // Create a blob container client that you use later to build an event processor client to receive and process events
       BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder()
           .connectionString(storageConnectionString)
           .containerName(storageContainerName)
           .buildAsyncClient();
   
       // Create a builder object that you will use later to build an event processor client to receive and process events and errors.
       EventProcessorClientBuilder eventProcessorClientBuilder = new EventProcessorClientBuilder()
           .connectionString(connectionString, eventHubName)
           .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME)
           .processEvent(PARTITION_PROCESSOR)
           .processError(ERROR_HANDLER)
           .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient));
   
       // Use the builder object to create an event processor client
       EventProcessorClient eventProcessorClient = eventProcessorClientBuilder.buildEventProcessorClient();
   
       System.out.println("Starting event processor");
       eventProcessorClient.start();
   
       System.out.println("Press enter to stop.");
       System.in.read();
   
       System.out.println("Stopping event processor");
       eventProcessorClient.stop();
       System.out.println("Event processor stopped.");
   
       System.out.println("Exiting process");
   }