Azure.Messaging.EventHubs (.NET)
SDK de streaming de eventos de alta taxa de transferência para enviar e receber eventos por meio do Azure Event Hubs.
Instalação
# Pacote principal (envio e recebimento simples)
dotnet add package Azure.Messaging.EventHubs
# Pacote do processador (recebimento em produção com checkpointing)
dotnet add package Azure.Messaging.EventHubs.Processor
# Autenticação
dotnet add package Azure.Identity
# Para checkpointing (necessário para EventProcessorClient)
dotnet add package Azure.Storage.Blobs
Versões Atuais: Azure.Messaging.EventHubs v5.12.2, Azure.Messaging.EventHubs.Processor v5.12.2
Variáveis de Ambiente
EVENTHUB_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net # Obrigatório: namespace totalmente qualificado do Event Hubs
EVENTHUB_NAME=<nome-do-eventhub> # Obrigatório: nome do Event Hub
BLOB_STORAGE_CONNECTION_STRING=<string-de-conexao-do-storage> # Alternativa à autenticação via Entra ID
BLOB_CONTAINER_NAME=<container-de-checkpoint> # Obrigatório: nome do container de checkpoint
EVENTHUB_CONNECTION_STRING=Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=... # Alternativa à autenticação via Entra ID
AZURE_TOKEN_CREDENTIALS=prod # Obrigatório apenas se DefaultAzureCredential for usado em produção
Autenticação
using Azure.Identity;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Producer;
// Desenvolvimento local: DefaultAzureCredential. Produção: defina AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=<credencial_especifica>
var credential = new DefaultAzureCredential(
DefaultAzureCredential.DefaultEnvironmentVariableName
);
// Ou use uma credencial específica diretamente em produção:
// Consulte 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);
Funções RBAC Necessárias:
- Envio:
Azure Event Hubs Data Sender - Recebimento:
Azure Event Hubs Data Receiver - Ambos:
Azure Event Hubs Data Owner
Tipos de Cliente
| Cliente | Finalidade | Quando Usar |
|---|---|---|
EventHubProducerClient | Enviar eventos imediatamente em lotes | Envio em tempo real, controle total sobre lotes |
EventHubBufferedProducerClient | Lotes automáticos com envio em segundo plano | Cenários de alto volume, do tipo "disparar e esquecer" |
EventHubConsumerClient | Leitura simples de eventos | Somente prototipagem, NÃO para produção |
EventProcessorClient | Processamento de eventos em produção | Sempre use este para recebimento em produção |
Fluxo de Trabalho Principal
1. Enviar Eventos (Lote)
using Azure.Identity;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Producer;
await using var producer = new EventHubProducerClient(
fullyQualifiedNamespace,
eventHubName,
new DefaultAzureCredential());
// Crie um lote (respeita limites de tamanho automaticamente)
using EventDataBatch batch = await producer.CreateBatchAsync();
// Adicione eventos ao lote
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))
{
// Lote está cheio - envie-o e crie um novo
await producer.SendAsync(batch);
batch = await producer.CreateBatchAsync();
if (!batch.TryAdd(eventData))
{
throw new Exception("Evento muito grande para um lote vazio");
}
}
}
// Envie eventos restantes
if (batch.Count > 0)
{
await producer.SendAsync(batch);
}
2. Enviar Eventos (Bufferizado - Alto Volume)
using Azure.Messaging.EventHubs.Producer;
var options = new EventHubBufferedProducerClientOptions
{
MaximumWaitTime = TimeSpan.FromSeconds(1)
};
await using var producer = new EventHubBufferedProducerClient(
fullyQualifiedNamespace,
eventHubName,
new DefaultAzureCredential(),
options);
// Lidar com sucesso/falha do envio
producer.SendEventBatchSucceededAsync += args =>
{
Console.WriteLine($"Lote enviado: {args.EventBatch.Count} eventos");
return Task.CompletedTask;
};
producer.SendEventBatchFailedAsync += args =>
{
Console.WriteLine($"Falha no lote: {args.Exception.Message}");
return Task.CompletedTask;
};
// Enfileire eventos (enviados automaticamente em segundo plano)
for (int i = 0; i < 1000; i++)
{
await producer.EnqueueEventAsync(new EventData($"Evento {i}"));
}
// Descarregue eventos restantes antes de descartar
await producer.FlushAsync();
3. Receber Eventos (Produção - EventProcessorClient)
using Azure.Identity;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Consumer;
using Azure.Messaging.EventHubs.Processor;
using Azure.Storage.Blobs;
// Container de blobs para checkpointing
var blobClient = new BlobContainerClient(
Environment.GetEnvironmentVariable("BLOB_STORAGE_CONNECTION_STRING"),
Environment.GetEnvironmentVariable("BLOB_CONTAINER_NAME"));
await blobClient.CreateIfNotExistsAsync();
// Crie o processador
var processor = new EventProcessorClient(
blobClient,
EventHubConsumerClient.DefaultConsumerGroup,
fullyQualifiedNamespace,
eventHubName,
new DefaultAzureCredential());
// Lidar com eventos
processor.ProcessEventAsync += async args =>
{
Console.WriteLine($"Partição: {args.Partition.PartitionId}");
Console.WriteLine($"Dados: {args.Data.EventBody}");
// Checkpoint após processamento (ou checkpoints em lote)
await args.UpdateCheckpointAsync();
};
// Lidar com erros
processor.ProcessErrorAsync += args =>
{
Console.WriteLine($"Erro: {args.Exception.Message}");
Console.WriteLine($"Partição: {args.PartitionId}");
return Task.CompletedTask;
};
// Iniciar processamento
await processor.StartProcessingAsync();
// Executar até ser cancelado
await Task.Delay(Timeout.Infinite, cancellationToken);
// Parar graciosamente
await processor.StopProcessingAsync();
4. Operações de Partição
// Obter IDs de partição
string[] partitionIds = await producer.GetPartitionIdsAsync();
// Enviar para partição específica (use com moderação)
var options = new SendEventOptions
{
PartitionId = "0"
};
await producer.SendAsync(events, options);
// Usar chave de partição (recomendado para ordenação)
var batchOptions = new CreateBatchOptions
{
PartitionKey = "cliente-123" // Eventos com a mesma chave vão para a mesma partição
};
using var batch = await producer.CreateBatchAsync(batchOptions);
Opções de EventPosition
Controle de onde começar a ler:
// Começar do início
EventPosition.Earliest
// Começar do final (novos eventos apenas)
EventPosition.Latest
// Começar de um offset específico
EventPosition.FromOffset(12345)
// Começar de um número de sequência específico
EventPosition.FromSequenceNumber(100)
// Começar de um horário específico
EventPosition.FromEnqueuedTime(DateTimeOffset.UtcNow.AddHours(-1))
Integração com 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());
});
// Injetar em controller/serviço
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);
}
}
Práticas Recomendadas
- Use
EventProcessorClientpara recebimento — Nunca useEventHubConsumerClientem produção - Faça checkpoint estrategicamente — Após N eventos ou intervalo de tempo, não a cada evento
- Use chaves de partição — Para garantias de ordenação dentro de uma partição
- Reutilize clientes — Crie uma vez, use como singleton (thread-safe)
- Use
await using— Garante descarte adequado - Lide com
ProcessErrorAsync— Sempre registre um manipulador de erros - Agrupe eventos — Use
CreateBatchAsync()para respeitar limites de tamanho - Use o produtor bufferizado — Para cenários de alto volume com lotes automáticos
Tratamento de Erros
using Azure.Messaging.EventHubs;
try
{
await producer.SendAsync(batch);
}
catch (EventHubsException ex) when (ex.Reason == EventHubsException.FailureReason.ServiceBusy)
{
// Tentar novamente com backoff
await Task.Delay(TimeSpan.FromSeconds(5));
}
catch (EventHubsException ex) when (ex.IsTransient)
{
// Erro transitório - seguro tentar novamente
Console.WriteLine($"Erro transitório: {ex.Message}");
}
catch (EventHubsException ex)
{
// Erro não transitório
Console.WriteLine($"Erro: {ex.Reason} - {ex.Message}");
}
Estratégias de Checkpointing
| Estratégia | Quando Usar |
|---|---|
| A cada evento | Volume baixo, dados críticos |
| A cada N eventos | Throughput/confiabilidade equilibrados |
| Baseada em tempo | Intervalos de checkpoint consistentes |
| Após conclusão do lote | Após processar um lote lógico |
// Checkpoint a cada 100 eventos
private int _eventCount = 0;
processor.ProcessEventAsync += async args =>
{
// Processar evento...
_eventCount++;
if (_eventCount >= 100)
{
await args.UpdateCheckpointAsync();
_eventCount = 0;
}
};
SDKs Relacionados
| SDK | Finalidade | Instalar |
|---|---|---|
Azure.Messaging.EventHubs | Envio/recebimento principal | dotnet add package Azure.Messaging.EventHubs |
Azure.Messaging.EventHubs.Processor | Processamento em produção | dotnet add package Azure.Messaging.EventHubs.Processor |
Azure.ResourceManager.EventHubs | Plano de gerenciamento (criar hubs) | dotnet add package Azure.ResourceManager.EventHubs |
Microsoft.Azure.WebJobs.Extensions.EventHubs | Binding do Azure Functions | dotnet add package Microsoft.Azure.WebJobs.Extensions.EventHubs |