Azure.Messaging.EventHubs (.NET)
高スループットのイベントストリーミングを実現し、Azure Event Hubs を介したイベントの送受信を行うための SDK です。
インストール
# コアパッケージ（送信とシンプルな受信）
dotnet add package Azure.Messaging.EventHubs
# プロセッサパッケージ（チェックポイント処理付きのプロダクション受信）
dotnet add package Azure.Messaging.EventHubs.Processor
# 認証
dotnet add package Azure.Identity
# チェックポイント処理用（EventProcessorClient に必要）
dotnet add package Azure.Storage.Blobs
現在のバージョン: Azure.Messaging.EventHubs v5.12.2、Azure.Messaging.EventHubs.Processor v5.12.2
環境変数
EVENTHUB_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net # 必須: Event Hubs の完全修飾名前空間
EVENTHUB_NAME=<event-hub-name> # 必須: Event Hub 名
BLOB_STORAGE_CONNECTION_STRING=<storage-connection-string> # Entra ID 認証の代替
BLOB_CONTAINER_NAME=<checkpoint-container> # 必須: チェックポイント用コンテナ名
EVENTHUB_CONNECTION_STRING=Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=... # Entra ID 認証の代替
AZURE_TOKEN_CREDENTIALS=prod # プロダクションで DefaultAzureCredential を使用する場合にのみ必要
認証
using Azure.Identity;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Producer;
// ローカル開発: DefaultAzureCredential。プロダクション: AZURE_TOKEN_CREDENTIALS=prod または AZURE_TOKEN_CREDENTIALS=<特定の資格情報>
var credential = new DefaultAzureCredential(
DefaultAzureCredential.DefaultEnvironmentVariableName
);
// またはプロダクションで特定の資格情報を直接使用:
// 詳細: https://learn.microsoft.com/dotnet/api/overview/azure/identity-readme?view=azure-dotnet#credential-classes
// var credential = new ManagedIdentityCredential();
var fullyQualifiedNamespace = Environment.GetEnvironmentVariable("EVENTHUB_FULLY_QUALIFIED_NAMESPACE");
var eventHubName = Environment.GetEnvironmentVariable("EVENTHUB_NAME");
var producer = new EventHubProducerClient(
fullyQualifiedNamespace,
eventHubName,
credential);
必要な RBAC ロール:
- 送信:
Azure Event Hubs Data Sender - 受信:
Azure Event Hubs Data Receiver - 両方:
Azure Event Hubs Data Owner
クライアントの種類
| クライアント | 目的 | 使用タイミング |
|---|---|---|
EventHubProducerClient | イベントをバッチで即時送信 | リアルタイム送信、バッチ処理を完全制御したい場合 |
EventHubBufferedProducerClient | バックグラウンド送信で自動バッチ処理 | 高負荷、fire-and-forget シナリオ |
EventHubConsumerClient | シンプルなイベント読み取り | プロトタイピングのみ、プロダクションでは非推奨 |
EventProcessorClient | プロダクション向けイベント処理 | プロダクションでの受信は常にこれを使用 |
コアワークフロー
1. イベントの送信（バッチ）
using Azure.Identity;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Producer;
await using var producer = new EventHubProducerClient(
fullyQualifiedNamespace,
eventHubName,
new DefaultAzureCredential());
// バッチを作成（サイズ制限を自動的に考慮）
using EventDataBatch batch = await producer.CreateBatchAsync();
// バッチにイベントを追加
var events = new[]
{
new EventData(BinaryData.FromString("{\"id\": 1, \"message\": \"Hello\"}")),
new EventData(BinaryData.FromString("{\"id\": 2, \"message\": \"World\"}"))
};
foreach (var eventData in events)
{
if (!batch.TryAdd(eventData))
{
// バッチがいっぱいになったので送信し、新しいバッチを作成
await producer.SendAsync(batch);
batch = await producer.CreateBatchAsync();
if (!batch.TryAdd(eventData))
{
throw new Exception("空のバッチに対してイベントが大きすぎます");
}
}
}
// 残りのイベントを送信
if (batch.Count > 0)
{
await producer.SendAsync(batch);
}
2. イベントの送信（バッファリング - 高負荷）
using Azure.Messaging.EventHubs.Producer;
var options = new EventHubBufferedProducerClientOptions
{
MaximumWaitTime = TimeSpan.FromSeconds(1)
};
await using var producer = new EventHubBufferedProducerClient(
fullyQualifiedNamespace,
eventHubName,
new DefaultAzureCredential(),
options);
// 送信成功/失敗のハンドリング
producer.SendEventBatchSucceededAsync += args =>
{
Console.WriteLine($"バッチ送信成功: {args.EventBatch.Count} 件のイベント");
return Task.CompletedTask;
};
producer.SendEventBatchFailedAsync += args =>
{
Console.WriteLine($"バッチ送信失敗: {args.Exception.Message}");
return Task.CompletedTask;
};
// イベントをキューに追加（バックグラウンドで自動送信）
for (int i = 0; i < 1000; i++)
{
await producer.EnqueueEventAsync(new EventData($"Event {i}"));
}
// 破棄前に残りのイベントをフラッシュ
await producer.FlushAsync();
3. イベントの受信（プロダクション - EventProcessorClient）
using Azure.Identity;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Consumer;
using Azure.Messaging.EventHubs.Processor;
using Azure.Storage.Blobs;
// チェックポイント用の BLOB コンテナ
var blobClient = new BlobContainerClient(
Environment.GetEnvironmentVariable("BLOB_STORAGE_CONNECTION_STRING"),
Environment.GetEnvironmentVariable("BLOB_CONTAINER_NAME"));
await blobClient.CreateIfNotExistsAsync();
// プロセッサを作成
var processor = new EventProcessorClient(
blobClient,
EventHubConsumerClient.DefaultConsumerGroup,
fullyQualifiedNamespace,
eventHubName,
new DefaultAzureCredential());
// イベントの処理
processor.ProcessEventAsync += async args =>
{
Console.WriteLine($"パーティション: {args.Partition.PartitionId}");
Console.WriteLine($"データ: {args.Data.EventBody}");
// 処理後にチェックポイント（またはバッチ単位でチェックポイント）
await args.UpdateCheckpointAsync();
};
// エラーの処理
processor.ProcessErrorAsync += args =>
{
Console.WriteLine($"エラー: {args.Exception.Message}");
Console.WriteLine($"パーティション: {args.PartitionId}");
return Task.CompletedTask;
};
// 処理を開始
await processor.StartProcessingAsync();
// キャンセルされるまで実行
await Task.Delay(Timeout.Infinite, cancellationToken);
// 正常に停止
await processor.StopProcessingAsync();
4. パーティション操作
// パーティション ID を取得
string[] partitionIds = await producer.GetPartitionIdsAsync();
// 特定のパーティションに送信（控えめに使用）
var options = new SendEventOptions
{
PartitionId = "0"
};
await producer.SendAsync(events, options);
// パーティションキーを使用（順序保証に推奨）
var batchOptions = new CreateBatchOptions
{
PartitionKey = "customer-123" // 同じキーのイベントは同じパーティションに送信される
};
using var batch = await producer.CreateBatchAsync(batchOptions);
EventPosition オプション
読み取り開始位置を制御:
// 最初から開始
EventPosition.Earliest
// 最後から開始（新しいイベントのみ）
EventPosition.Latest
// 特定のオフセットから開始
EventPosition.FromOffset(12345)
// 特定のシーケンス番号から開始
EventPosition.FromSequenceNumber(100)
// 特定の時刻から開始
EventPosition.FromEnqueuedTime(DateTimeOffset.UtcNow.AddHours(-1))
ASP.NET Core 統合
// Program.cs
using Azure.Identity;
using Azure.Messaging.EventHubs.Producer;
using Microsoft.Extensions.Azure;
builder.Services.AddAzureClients(clientBuilder =>
{
clientBuilder.AddEventHubProducerClient(
builder.Configuration["EventHub:FullyQualifiedNamespace"],
builder.Configuration["EventHub:Name"]);
clientBuilder.UseCredential(new DefaultAzureCredential());
});
// コントローラー/サービスで依存性注入
public class EventService
{
private readonly EventHubProducerClient _producer;
public EventService(EventHubProducerClient producer)
{
_producer = producer;
}
public async Task SendAsync(string message)
{
using var batch = await _producer.CreateBatchAsync();
batch.TryAdd(new EventData(message));
await _producer.SendAsync(batch);
}
}
ベストプラクティス
- 受信には
EventProcessorClientを使用 — プロダクションでEventHubConsumerClientを使用しない - チェックポイントは戦略的に — すべてのイベントではなく、N イベント後または時間間隔で実施
- パーティションキーを使用 — パーティション内の順序保証のため
- クライアントを再利用 — 一度作成してシングルトンとして使用（スレッドセーフ）
await usingを使用 — 適切な破棄を保証ProcessErrorAsyncを処理 — 常にエラーハンドラを登録- イベントをバッチ処理 —
CreateBatchAsync()を使用してサイズ制限を尊重 - バッファリングプロデューサを使用 — 自動バッチ処理を伴う高負荷シナリオ向け
エラーハンドリング
using Azure.Messaging.EventHubs;
try
{
await producer.SendAsync(batch);
}
catch (EventHubsException ex) when (ex.Reason == EventHubsException.FailureReason.ServiceBusy)
{
// バックオフしてリトライ
await Task.Delay(TimeSpan.FromSeconds(5));
}
catch (EventHubsException ex) when (ex.IsTransient)
{
// 一時的エラー - リトライしても安全
Console.WriteLine($"一時的エラー: {ex.Message}");
}
catch (EventHubsException ex)
{
// 非一時的エラー
Console.WriteLine($"エラー: {ex.Reason} - {ex.Message}");
}
チェックポイント戦略
| 戦略 | 使用タイミング |
|---|---|
| すべてのイベント | 低負荷、重要なデータ |
| N イベントごと | スループットと信頼性のバランス |
| 時間ベース | 一貫したチェックポイント間隔 |
| バッチ完了時 | 論理的なバッチ処理後 |
// 100 イベントごとにチェックポイント
private int _eventCount = 0;
processor.ProcessEventAsync += async args =>
{
// イベントを処理...
_eventCount++;
if (_eventCount >= 100)
{
await args.UpdateCheckpointAsync();
_eventCount = 0;
}
};
関連 SDK
| SDK | 目的 | インストール |
|---|---|---|
Azure.Messaging.EventHubs | コアの送受信 | dotnet add package Azure.Messaging.EventHubs |
Azure.Messaging.EventHubs.Processor | プロダクション処理 | dotnet add package Azure.Messaging.EventHubs.Processor |
Azure.ResourceManager.EventHubs | 管理プレーン（ハブの作成） | dotnet add package Azure.ResourceManager.EventHubs |
Microsoft.Azure.WebJobs.Extensions.EventHubs | Azure Functions バインディング | dotnet add package Microsoft.Azure.WebJobs.Extensions.EventHubs |