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 | 백그라운드 전송으로 자동 배치 | 대용량, 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 |