クイック スタート: .NET を使用して Azure Event Hubs との間でイベントを送受信する

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

ローカルで開発するときは、Azure Event Hubs に接続するユーザー アカウントに適切なアクセス許可があることを確認します。 メッセージを送受信するには、 Azure Event Hubs データ所有者 ロールが必要です。 このロールを自分に割り当てるには、ユーザー アクセス管理者ロール、または Microsoft.Authorization/roleAssignments/write アクションを含む別のロールが必要です。 Azure portal、Azure CLI、または Azure PowerShell を使用して、ユーザーに Azure RBAC ロールを割り当てることができます。 詳細については、「 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 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>

Visual Studio を起動して Azure にサインインする

次の手順を使用して、Service Bus 名前空間へのアクセスを承認できます。

1. Visual Studio を起動します。 [作業の開始] ウィンドウが表示されたら、右側のウィンドウで [コードなしで続行] リンクを選択します。

2. Visual Studio の右上隅にある [サインイン] ボタンを選びます。

   

3. 以前にロールを割り当てた Microsoft Entra アカウントを使用してサインインします。

   

接続文字列を取得する

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

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

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

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

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

   

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

1. メニューから [ツール]>[NuGet パッケージ マネージャー]>[パッケージ マネージャー コンソール] の順に選択します。

2. 次のコマンドを実行して Azure.Messaging.EventHubs と Azure.Identity NuGet パッケージをインストールします。 Enter キーを押して 2 つ目のコマンドを実行します。

   Install-Package Azure.Messaging.EventHubs
   Install-Package Azure.Identity
   

1. メニューから [ツール]>[NuGet パッケージ マネージャー]>[パッケージ マネージャー コンソール] の順に選択します。

2. 次のコマンドを実行して、Azure.Messaging.EventHubs NuGet パッケージをインストールします。

   Install-Package Azure.Messaging.EventHubs
   

1. Program.cs ファイル内の既存のコードを、次のサンプル コードに置き換えます。 その後、<EVENT_HUB_NAMESPACE> パラメーターの <HUB_NAME> および EventHubProducerClient プレースホルダーの値を、Event Hubs 名前空間とイベント ハブの名前に置き換えます。 たとえば、"spehubns0309.servicebus.windows.net" と "spehub" などです。

   コードの重要な手順を次に示します。

   1. 名前空間と Event Hub 名を使用して EventHubProducerClient オブジェクトを作成します。
   2. EventHubProducerClient オブジェクトで CreateBatchAsync メソッドを呼び出して、EventDataBatch オブジェクトを作成します。
   3. EventDataBatch.TryAdd メソッドを使って、バッチにイベントを追加します。
   4. EventHubProducerClient.SendAsync メソッドを使って、メッセージのバッチをイベント ハブに送信します。

   using Azure.Identity;
   using Azure.Messaging.EventHubs;
   using Azure.Messaging.EventHubs.Producer;
   using System.Text;
   
   // number of events to be sent to the event hub
   int numOfEvents = 3;
   
   // The Event Hubs client types are safe to cache and use as a singleton for the lifetime
   // of the application, which is best practice when events are being published or read regularly.
   // TODO: Replace the <EVENT_HUB_NAMESPACE> and <HUB_NAME> placeholder values
   EventHubProducerClient producerClient = new EventHubProducerClient(
       "<EVENT_HUB_NAMESPACE>.servicebus.windows.net",
       "<HUB_NAME>",
       new DefaultAzureCredential());
   
   // Create a batch of events 
   using EventDataBatch eventBatch = await producerClient.CreateBatchAsync();
   
   for (int i = 1; i <= numOfEvents; i++)
   {
       if (!eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes($"Event {i}"))))
       {
           // if it is too large for the batch
           throw new Exception($"Event {i} is too large for the batch and cannot be sent.");
       }
   }
   
   try
   {
       // Use the producer client to send the batch of events to the event hub
       await producerClient.SendAsync(eventBatch);
       Console.WriteLine($"A batch of {numOfEvents} events has been published.");
       Console.ReadLine();
   }
   finally
   {
       await producerClient.DisposeAsync();
   }
   

1. Program.cs ファイル内の既存のコードを、次のサンプル コードに置き換えます。 その後、<CONNECTION_STRING> パラメーターの <HUB_NAME> および EventHubProducerClient プレースホルダーの値を置き換えます。

   コードの重要な手順を次に示します。

   1. 名前空間へのプライマリ接続文字列とイベント ハブの名前を使って、EventHubProducerClient オブジェクトを作成します。
   2. EventHubProducerClient オブジェクトで CreateBatchAsync メソッドを呼び出して、EventDataBatch オブジェクトを作成します。
   3. EventDataBatch.TryAdd メソッドを使って、バッチにイベントを追加します。
   4. EventHubProducerClient.SendAsync メソッドを使って、メッセージのバッチをイベント ハブに送信します。

   using Azure.Messaging.EventHubs;
   using Azure.Messaging.EventHubs.Producer;
   using System.Text;
   
   // number of events to be sent to the event hub
   int numOfEvents = 3;
   
   // The Event Hubs client types are safe to cache and use as a singleton for the lifetime
   // of the application, which is best practice when events are being published or read regularly.
   // TODO: Replace the <CONNECTION_STRING> and <HUB_NAME> placeholder values
   EventHubProducerClient producerClient = new EventHubProducerClient(
       "<CONNECTION_STRING>",
       "<HUB_NAME>");
   
   // Create a batch of events 
   using EventDataBatch eventBatch = await producerClient.CreateBatchAsync();
   
   for (int i = 1; i <= numOfEvents; i++)
   {
       if (!eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes($"Event {i}"))))
       {
           // if it is too large for the batch
           throw new Exception($"Event {i} is too large for the batch and cannot be sent.");
       }
   }
   
   try
   {
       // Use the producer client to send the batch of events to the event hub
       await producerClient.SendAsync(eventBatch);
       Console.WriteLine($"A batch of {numOfEvents} events has been published.");
       Console.ReadLine();
   }
   finally
   {
       await producerClient.DisposeAsync();
   }
   

ローカルで開発する場合は、BLOB データにアクセスするユーザー アカウントに適切なアクセス許可があることを確認します。 BLOB データの読み取りと書き込みを行うには、 ストレージ BLOB データ共同作成者 が必要です。 このロールを自分に割り当てるには、 ユーザー アクセス管理者 ロール、または Microsoft.Authorization/roleAssignments/write アクションを含む別のロールを割り当てる必要があります。 Azure portal、Azure CLI、または Azure PowerShell を使用して、ユーザーに Azure RBAC ロールを割り当てることができます。 詳しくは、「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>

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

接続文字列とコンテナー名を書き留めておきます。 イベント ハブからイベントを受信するには、コード内でそれらを使用します。

1. メニューから [ツール]>[NuGet パッケージ マネージャー]>[パッケージ マネージャー コンソール] の順に選択します。

2. [パッケージ マネージャー コンソール] ウィンドウで、 [既定のプロジェクト] に [EventHubsReceiver] が選択されていることを確認します。 そうでない場合は、ドロップダウン リストを使用して [EventHubsReceiver] を選択します。

3. 次のコマンドを実行して Azure.Messaging.EventHubs と Azure.Identity NuGet パッケージをインストールします。 Enter キーを押して最後のコマンドを実行します。

   Install-Package Azure.Messaging.EventHubs
   Install-Package Azure.Messaging.EventHubs.Processor
   Install-Package Azure.Identity
   

1. メニューから [ツール]>[NuGet パッケージ マネージャー]>[パッケージ マネージャー コンソール] の順に選択します。

2. [パッケージ マネージャー コンソール] ウィンドウで、 [既定のプロジェクト] に [EventHubsReceiver] が選択されていることを確認します。 そうでない場合は、ドロップダウン リストを使用して [EventHubsReceiver] を選択します。

3. 次のコマンドを実行して、Azure.Messaging.EventHubs NuGet パッケージをインストールします。

   Install-Package Azure.Messaging.EventHubs
   Install-Package Azure.Messaging.EventHubs.Processor
   

1. Program.cs ファイル内の既存のコードを、次のサンプル コードに置き換えます。 その後、<STORAGE_ACCOUNT_NAME> URI の <BLOB_CONTAINER_NAME> および BlobContainerClient プレースホルダーの値を置き換えます。 <EVENT_HUB_NAMESPACE> の <HUB_NAME> プレースホルダーと EventProcessorClient プレースホルダーの値も置き換えてください。

   コードの重要な手順を次に示します。

   1. Event Hubs 名前空間とイベント ハブ名を使用して、EventProcessorClient オブジェクトを作成します。 前に作成した Azure Storage でコンテナー用の BlobContainerClient オブジェクトを作成する必要があります。
   2. Eventprocessorclient オブジェクトの ProcessEventAsync および ProcessErrorAsync イベント用のハンドラーを指定します。
   3. EventProcessorClient オブジェクトで StartProcessingAsync を呼び出して、イベントの処理を開始します。
   4. EventProcessorClient オブジェクトで StopProcessingAsync を呼び出して、30 秒後にイベントの処理を停止します。

   using Azure.Identity;
   using Azure.Messaging.EventHubs;
   using Azure.Messaging.EventHubs.Consumer;
   using Azure.Messaging.EventHubs.Processor;
   using Azure.Storage.Blobs;
   using System.Text;
   
   // Create a blob container client that the event processor will use
   // TODO: Replace <STORAGE_ACCOUNT_NAME> and <BLOB_CONTAINER_NAME> with actual names
   BlobContainerClient storageClient = new BlobContainerClient(
       new Uri("https://<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/<BLOB_CONTAINER_NAME>"),
       new DefaultAzureCredential());
   
   // Create an event processor client to process events in the event hub
   // TODO: Replace the <EVENT_HUBS_NAMESPACE> and <HUB_NAME> placeholder values
   var processor = new EventProcessorClient(
       storageClient,
       EventHubConsumerClient.DefaultConsumerGroupName,
       "<EVENT_HUB_NAMESPACE>.servicebus.windows.net",
       "<HUB_NAME>",
       new DefaultAzureCredential());
   
   // Register handlers for processing events and handling errors
   processor.ProcessEventAsync += ProcessEventHandler;
   processor.ProcessErrorAsync += ProcessErrorHandler;
   
   // Start the processing
   await processor.StartProcessingAsync();
   
   // Wait for 30 seconds for the events to be processed
   await Task.Delay(TimeSpan.FromSeconds(30));
   
   // Stop the processing
   await processor.StopProcessingAsync();
   
   Task ProcessEventHandler(ProcessEventArgs eventArgs)
   {
       // Write the body of the event to the console window
       Console.WriteLine("\tReceived event: {0}", Encoding.UTF8.GetString(eventArgs.Data.Body.ToArray()));
       Console.ReadLine();
       return Task.CompletedTask;
   }
   
   Task ProcessErrorHandler(ProcessErrorEventArgs eventArgs)
   {
       // Write details about the error to the console window
       Console.WriteLine($"\tPartition '{eventArgs.PartitionId}': an unhandled exception was encountered. This was not expected to happen.");
       Console.WriteLine(eventArgs.Exception.Message);
       Console.ReadLine();
       return Task.CompletedTask;
   }
   

1. Program.cs ファイル内の既存のコードを、次のサンプル コードに置き換えます。 その後、<AZURE_STORAGE_CONNECTION_STRING> URI の <BLOB_CONTAINER_NAME> および BlobContainerClient プレースホルダーの値を置き換えます。 <EVENT_HUB_NAMESPACE_CONNECTION_STRING> の <HUB_NAME> プレースホルダーと EventProcessorClient プレースホルダーの値も置き換えてください。

   コードの重要な手順を次に示します。

   1. 名前空間へのプライマリ接続文字列とイベント ハブを使って、EventProcessorClient オブジェクトを作成します。 前に作成した Azure Storage でコンテナー用の BlobContainerClient オブジェクトを作成する必要があります。
   2. Eventprocessorclient オブジェクトの ProcessEventAsync および ProcessErrorAsync イベント用のハンドラーを指定します。
   3. EventProcessorClient オブジェクトで StartProcessingAsync を呼び出して、イベントの処理を開始します。
   4. EventProcessorClient オブジェクトで StopProcessingAsync を呼び出して、30 秒後にイベントの処理を停止します。

   using Azure.Messaging.EventHubs;
   using Azure.Messaging.EventHubs.Consumer;
   using Azure.Messaging.EventHubs.Processor;
   using Azure.Storage.Blobs;
   using System.Text;
   
   // Create a blob container client that the event processor will use 
   BlobContainerClient storageClient = new BlobContainerClient(
       "<AZURE_STORAGE_CONNECTION_STRING>", "<BLOB_CONTAINER_NAME>");
   
   // Create an event processor client to process events in the event hub
   var processor = new EventProcessorClient(
       storageClient,
       EventHubConsumerClient.DefaultConsumerGroupName,
       "<EVENT_HUBS_NAMESPACE_CONNECTION_STRING>",
       "<HUB_NAME>");
   
   // Register handlers for processing events and handling errors
   processor.ProcessEventAsync += ProcessEventHandler;
   processor.ProcessErrorAsync += ProcessErrorHandler;
   
   // Start the processing
   await processor.StartProcessingAsync();
   
   // Wait for 30 seconds for the events to be processed
   await Task.Delay(TimeSpan.FromSeconds(30));
   
   // Stop the processing
   await processor.StopProcessingAsync();
   
   Task ProcessEventHandler(ProcessEventArgs eventArgs)
   {
       // Write the body of the event to the console window
       Console.WriteLine("\tReceived event: {0}", Encoding.UTF8.GetString(eventArgs.Data.Body.ToArray()));
       return Task.CompletedTask;
   }
   
   Task ProcessErrorHandler(ProcessErrorEventArgs eventArgs)
   {
       // Write details about the error to the console window
       Console.WriteLine($"\tPartition '{eventArgs.PartitionId}': an unhandled exception was encountered. This was not expected to happen.");
       Console.WriteLine(eventArgs.Exception.Message);
       return Task.CompletedTask;
   }