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 代码片段需要已验证的安装版本和测试样例。将事件加入缓冲区不代表代理已确认；请明确处理成功或失败，并在关闭时刷新缓冲区。接收采用至少一次传递：先执行持久化的幂等处理，再设置分区内检查点；并发处理多个分区时，切勿将下方的共享计数器用作正确性边界。解引用空读取前，请先检查是否确实存在事件。将处理生命周期封装在 try/finally 中，确保即使发生取消也会调用 StopProcessingAsync；请妥善保管事件正文和错误信息。可以同时授予 Sender 和 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

适用场景

此技能适用于执行概览中描述的工作流或操作。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为特定环境验证、测试或专家评审的替代品。
• 如果缺少必需的输入、权限、安全边界或成功标准，请停止并请求澄清。