microsoft/azure-eventhub-dotnet

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=<specific_credential>
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	自動批次處理與背景傳送	高容量、發送後即忘的案例
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. 分割區操作

// 取得分割區標識符
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);
    }
}

最佳做法

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)
{
    // 使用指數退避重試
    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