azure-eventhub-dotnet

Azure.Messaging.EventHubs（.NET）

Azure Event Hubs を介してイベントを送受信するための、高スループットのイベントストリーミング SDK です。

インストール

# Core package (sending and simple receiving)
dotnet add package Azure.Messaging.EventHubs

# Processor package (production receiving with checkpointing)
dotnet add package Azure.Messaging.EventHubs.Processor

# Authentication
dotnet add package Azure.Identity

# For checkpointing (required by 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
EVENTHUB_NAME=<event-hub-name>

# For checkpointing (EventProcessorClient)
BLOB_STORAGE_CONNECTION_STRING=<storage-connection-string>
BLOB_CONTAINER_NAME=<checkpoint-container>

# Alternative: Connection string auth (not recommended for production)
EVENTHUB_CONNECTION_STRING=Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=...

認証

using Azure.Identity;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Producer;

// Always use DefaultAzureCredential for production
var credential = new DefaultAzureCredential();

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

配信、チェックポイント、シャットダウンの制限

これらの SDK コードは、インストール済みバージョンとテスト用フィクスチャの確認が必要です。バッファーにイベントを追加しても、ブローカーによる確認応答にはなりません。成功／失敗を処理し、シャットダウン時のフラッシュを明示的に行ってください。受信は少なくとも 1 回の配信です。パーティションごとのチェックポイントを設定する前に、永続的で冪等な処理を行ってください。以下の共有カウンターを、並行するパーティション間で正確性を担保する境界として使用しないでください。読み取り結果が空の場合は、実際のイベントがあることを確認してから参照してください。処理のライフタイムを try/finally で囲み、キャンセル時にも StopProcessingAsync が呼び出されるようにしてください。イベント本文とエラーは非公開にしてください。Data Sender ロールと Data Receiver ロールを組み合わせれば、Data Owner を付与せずに送受信できます。認証情報とテナントを意図的に選び、作成、送信、受信、チェックポイントストレージの各権限の範囲を設定してください。

クライアントの種類

クライアント	用途	使用する場面
EventHubProducerClient	イベントをバッチで即時送信	リアルタイム送信、バッチ処理の完全な制御
EventHubBufferedProducerClient	バックグラウンドで自動的にバッチ送信	大量データの送信、結果を待たないシナリオ
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());

// Create a batch (respects size limits automatically)
EventDataBatch batch = await producer.CreateBatchAsync();

// Add events to batch
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))
    {
        // Batch is full - send it and create a new one
        await producer.SendAsync(batch);
        batch.Dispose();
        batch = await producer.CreateBatchAsync();
        
        if (!batch.TryAdd(eventData))
        {
            throw new Exception("Event too large for empty batch");
        }
    }
}

// Send remaining events; production code wraps the whole loop in try/finally
// so the current batch is disposed on every exception path.
if (batch.Count > 0)
{
    await producer.SendAsync(batch);
}
batch.Dispose();

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);

// Handle send success/failure
producer.SendEventBatchSucceededAsync += args =>
{
    Console.WriteLine($"Batch sent: {args.EventBatch.Count} events");
    return Task.CompletedTask;
};

producer.SendEventBatchFailedAsync += args =>
{
    Console.WriteLine($"Batch failed: {args.Exception.Message}");
    return Task.CompletedTask;
};

// Enqueue events (sent automatically in background)
for (int i = 0; i < 1000; i++)
{
    await producer.EnqueueEventAsync(new EventData($"Event {i}"));
}

// Flush remaining events before disposing
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 container for checkpointing
var blobClient = new BlobContainerClient(
    Environment.GetEnvironmentVariable("BLOB_STORAGE_CONNECTION_STRING"),
    Environment.GetEnvironmentVariable("BLOB_CONTAINER_NAME"));

await blobClient.CreateIfNotExistsAsync();

// Create processor
var processor = new EventProcessorClient(
    blobClient,
    EventHubConsumerClient.DefaultConsumerGroup,
    fullyQualifiedNamespace,
    eventHubName,
    new DefaultAzureCredential());

// Handle events
processor.ProcessEventAsync += async args =>
{
    Console.WriteLine($"Partition: {args.Partition.PartitionId}");
    Console.WriteLine($"Data: {args.Data.EventBody}");
    
    // Checkpoint after processing (or batch checkpoints)
    await args.UpdateCheckpointAsync();
};

// Handle errors
processor.ProcessErrorAsync += args =>
{
    Console.WriteLine($"Error: {args.Exception.Message}");
    Console.WriteLine($"Partition: {args.PartitionId}");
    return Task.CompletedTask;
};

// Start processing
await processor.StartProcessingAsync();

// Run until cancelled
await Task.Delay(Timeout.Infinite, cancellationToken);

// Stop gracefully
await processor.StopProcessingAsync();

4. パーティション操作

// Get partition IDs
string[] partitionIds = await producer.GetPartitionIdsAsync();

// Send to specific partition (use sparingly)
var options = new SendEventOptions
{
    PartitionId = "0"
};
await producer.SendAsync(events, options);

// Use partition key (recommended for ordering)
var batchOptions = new CreateBatchOptions
{
    PartitionKey = "customer-123"  // Events with same key go to same partition
};
using var batch = await producer.CreateBatchAsync(batchOptions);

EventPosition のオプション

読み取り開始位置を指定します。

// Start from beginning
EventPosition.Earliest

// Start from end (new events only)
EventPosition.Latest

// Start from specific offset
EventPosition.FromOffset(12345)

// Start from specific sequence number
EventPosition.FromSequenceNumber(100)

// Start from specific time
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());
});

// Inject in controller/service
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();
        if (!batch.TryAdd(new EventData(message)))
            throw new ArgumentException("Message exceeds the empty batch limit", nameof(message));
        await _producer.SendAsync(batch);
    }
}

ベストプラクティス

1. 受信には EventProcessorClient を使用する — 本番環境では EventHubConsumerClient を使用しないでください
2. チェックポイントを戦略的に設定する — すべてのイベントごとではなく、N 件ごと、または一定の時間間隔ごとに設定します
3. パーティションキーを使用する — パーティション内の順序を保証します
4. クライアントを再利用する — 一度作成し、シングルトンとして使用します（スレッドセーフです）
5. await using を使用する — 適切に破棄されるようにします
6. ProcessErrorAsync を処理する — エラーハンドラーを必ず登録します
7. イベントをバッチ処理する — サイズ制限を守るために CreateBatchAsync() を使用します
8. バッファー処理プロデューサーを使用する — 自動バッチ処理を行う大量データのシナリオに適しています

エラー処理

using Azure.Messaging.EventHubs;

try
{
    await producer.SendAsync(batch);
}
catch (EventHubsException ex) when (ex.Reason == EventHubsException.FailureReason.ServiceBusy)
{
    // Retry with backoff
    await Task.Delay(TimeSpan.FromSeconds(5));
}
catch (EventHubsException ex) when (ex.IsTransient)
{
    // Transient error - safe to retry
    Console.WriteLine($"Transient error: {ex.Message}");
}
catch (EventHubsException ex)
{
    // Non-transient error
    Console.WriteLine($"Error: {ex.Reason} - {ex.Message}");
}

チェックポイント戦略

戦略	使用する場面
イベントごと	データ量が少なく、データが重要な場合
N 件ごと	スループットと信頼性のバランスを取る場合
時間ベース	一定の間隔でチェックポイントを設定する場合
バッチ完了時	論理的なバッチの処理後

// Checkpoint every 100 events
private int _eventCount = 0;

processor.ProcessEventAsync += async args =>
{
    // Process event...
    
    _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

使用する場面

このスキルは、概要で説明されているワークフローやアクションを実行する場合に適用できます。

制限事項

• このスキルは、タスクが上記のスコープに明確に一致する場合にのみ使用してください。
• 出力を、環境固有の検証、テスト、専門家によるレビューの代替として扱わないでください。
• 必要な入力、権限、安全境界、成功基準が欠けている場合は、作業を中断して確認を求めてください。