Azure.Messaging.EventHubs (.NET)
用于通过 Azure 事件中心发送和接收事件的高吞吐量事件流 SDK。
安装
# 核心包（发送和简单接收）
dotnet add package Azure.Messaging.EventHubs
# Processor 包（带检查点的生产环境接收）
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 # 必需：事件中心完全限定命名空间
EVENTHUB_NAME=<event-hub-name> # 必需：事件中心名称
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. 分区操作
// 获取分区 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 |