microsoft/azure-eventhub-dotnet

Azure.Messaging.EventHubs (.NET)

SDK de transmisión de eventos de alto rendimiento para enviar y recibir eventos mediante Azure Event Hubs.

Instalación

# Paquete principal (envío y recepción simple)
dotnet add package Azure.Messaging.EventHubs

# Paquete de procesador (recepción en producción con checkpointing)
dotnet add package Azure.Messaging.EventHubs.Processor

# Autenticación
dotnet add package Azure.Identity

# Para checkpointing (requerido por EventProcessorClient)
dotnet add package Azure.Storage.Blobs

Versiones actuales: Azure.Messaging.EventHubs v5.12.2, Azure.Messaging.EventHubs.Processor v5.12.2

Variables de entorno

EVENTHUB_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net  # Requerido: Espacio de nombres completo de Event Hubs
EVENTHUB_NAME=<event-hub-name>  # Requerido: Nombre del Event Hub
BLOB_STORAGE_CONNECTION_STRING=<storage-connection-string>  # Alternativa a la autenticación de Entra ID
BLOB_CONTAINER_NAME=<checkpoint-container>  # Requerido: Nombre del contenedor de puntos de control
EVENTHUB_CONNECTION_STRING=Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=...  # Alternativa a la autenticación de Entra ID
AZURE_TOKEN_CREDENTIALS=prod  # Requerido solo si se usa DefaultAzureCredential en producción

Autenticación

using Azure.Identity;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Producer;

// Desarrollo local: DefaultAzureCredential. Producción: establezca AZURE_TOKEN_CREDENTIALS=prod o AZURE_TOKEN_CREDENTIALS=<credential_especifica>
var credential = new DefaultAzureCredential(
    DefaultAzureCredential.DefaultEnvironmentVariableName
);
// O use una credencial específica directamente en producción:
// 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);

Roles de RBAC requeridos:

• Envío: Azure Event Hubs Data Sender
• Recepción: Azure Event Hubs Data Receiver
• Ambos: Azure Event Hubs Data Owner

Tipos de cliente

Cliente	Propósito	Cuándo usarlo
EventHubProducerClient	Enviar eventos inmediatamente en lotes	Envío en tiempo real, control total sobre los lotes
EventHubBufferedProducerClient	Agrupación automática con envío en segundo plano	Escenarios de alto volumen y de tipo "disparar y olvidar"
EventHubConsumerClient	Lectura simple de eventos	Solo para prototipos, NO para producción
EventProcessorClient	Procesamiento de eventos en producción	Úselo siempre para recepción en producción

Flujo de trabajo 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());

// Crear un lote (respeta los límites de tamaño automáticamente)
using EventDataBatch batch = await producer.CreateBatchAsync();

// Agregar eventos al lote
var events = new[]
{
    new EventData(BinaryData.FromString("{\"id\": 1, \"message\": \"Hola\"}")),
    new EventData(BinaryData.FromString("{\"id\": 2, \"message\": \"Mundo\"}"))
};

foreach (var eventData in events)
{
    if (!batch.TryAdd(eventData))
    {
        // El lote está lleno - enviarlo y crear uno nuevo
        await producer.SendAsync(batch);
        batch = await producer.CreateBatchAsync();
        
        if (!batch.TryAdd(eventData))
        {
            throw new Exception("El evento es demasiado grande para un lote vacío");
        }
    }
}

// Enviar eventos restantes
if (batch.Count > 0)
{
    await producer.SendAsync(batch);
}

2. Enviar eventos (almacenado en búfer - Alto volumen)

using Azure.Messaging.EventHubs.Producer;

var options = new EventHubBufferedProducerClientOptions
{
    MaximumWaitTime = TimeSpan.FromSeconds(1)
};

await using var producer = new EventHubBufferedProducerClient(
    fullyQualifiedNamespace,
    eventHubName,
    new DefaultAzureCredential(),
    options);

// Manejar éxito/fallo del envío
producer.SendEventBatchSucceededAsync += args =>
{
    Console.WriteLine($"Lote enviado: {args.EventBatch.Count} eventos");
    return Task.CompletedTask;
};

producer.SendEventBatchFailedAsync += args =>
{
    Console.WriteLine($"Fallo del lote: {args.Exception.Message}");
    return Task.CompletedTask;
};

// Poner eventos en cola (se envían automáticamente en segundo plano)
for (int i = 0; i < 1000; i++)
{
    await producer.EnqueueEventAsync(new EventData($"Evento {i}"));
}

// Vaciar eventos restantes antes de desechar
await producer.FlushAsync();

3. Recibir eventos (Producción - EventProcessorClient)

using Azure.Identity;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Consumer;
using Azure.Messaging.EventHubs.Processor;
using Azure.Storage.Blobs;

// Contenedor de blobs para puntos de control
var blobClient = new BlobContainerClient(
    Environment.GetEnvironmentVariable("BLOB_STORAGE_CONNECTION_STRING"),
    Environment.GetEnvironmentVariable("BLOB_CONTAINER_NAME"));

await blobClient.CreateIfNotExistsAsync();

// Crear procesador
var processor = new EventProcessorClient(
    blobClient,
    EventHubConsumerClient.DefaultConsumerGroup,
    fullyQualifiedNamespace,
    eventHubName,
    new DefaultAzureCredential());

// Manejar eventos
processor.ProcessEventAsync += async args =>
{
    Console.WriteLine($"Partición: {args.Partition.PartitionId}");
    Console.WriteLine($"Datos: {args.Data.EventBody}");
    
    // Punto de control después del procesamiento (o puntos de control por lotes)
    await args.UpdateCheckpointAsync();
};

// Manejar errores
processor.ProcessErrorAsync += args =>
{
    Console.WriteLine($"Error: {args.Exception.Message}");
    Console.WriteLine($"Partición: {args.PartitionId}");
    return Task.CompletedTask;
};

// Iniciar procesamiento
await processor.StartProcessingAsync();

// Ejecutar hasta que se cancele
await Task.Delay(Timeout.Infinite, cancellationToken);

// Detener correctamente
await processor.StopProcessingAsync();

4. Operaciones de partición

// Obtener IDs de partición
string[] partitionIds = await producer.GetPartitionIdsAsync();

// Enviar a una partición específica (usar con moderación)
var options = new SendEventOptions
{
    PartitionId = "0"
};
await producer.SendAsync(events, options);

// Usar clave de partición (recomendado para orden)
var batchOptions = new CreateBatchOptions
{
    PartitionKey = "cliente-123"  // Los eventos con la misma clave van a la misma partición
};
using var batch = await producer.CreateBatchAsync(batchOptions);

Opciones de EventPosition

Controlar desde dónde empezar a leer:

// Empezar desde el principio
EventPosition.Earliest

// Empezar desde el final (solo eventos nuevos)
EventPosition.Latest

// Empezar desde un desplazamiento específico
EventPosition.FromOffset(12345)

// Empezar desde un número de secuencia específico
EventPosition.FromSequenceNumber(100)

// Empezar desde una hora específica
EventPosition.FromEnqueuedTime(DateTimeOffset.UtcNow.AddHours(-1))

Integración con 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());
});

// Inyectar en el controlador/servicio
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);
    }
}

Mejores prácticas

1. Use EventProcessorClient para recepción — Nunca use EventHubConsumerClient en producción
2. Puntos de control estratégicos — Después de N eventos o un intervalo de tiempo, no en cada evento
3. Use claves de partición — Para garantías de orden dentro de una partición
4. Reutilice clientes — Cree una vez, use como singleton (seguro para subprocesos)
5. Use await using — Asegura una eliminación adecuada
6. Maneje ProcessErrorAsync — Siempre registre un controlador de errores
7. Agrupe eventos en lotes — Use CreateBatchAsync() para respetar los límites de tamaño
8. Use el productor almacenado en búfer — Para escenarios de alto volumen con agrupación automática

Manejo de errores

using Azure.Messaging.EventHubs;

try
{
    await producer.SendAsync(batch);
}
catch (EventHubsException ex) when (ex.Reason == EventHubsException.FailureReason.ServiceBusy)
{
    // Reintentar con retroceso
    await Task.Delay(TimeSpan.FromSeconds(5));
}
catch (EventHubsException ex) when (ex.IsTransient)
{
    // Error transitorio - seguro de reintentar
    Console.WriteLine($"Error transitorio: {ex.Message}");
}
catch (EventHubsException ex)
{
    // Error no transitorio
    Console.WriteLine($"Error: {ex.Reason} - {ex.Message}");
}

Estrategias de checkpointing

Estrategia	Cuándo usarla
Cada evento	Bajo volumen, datos críticos
Cada N eventos	Rendimiento/fiabilidad equilibrados
Basada en tiempo	Intervalos de punto de control consistentes
Finalización del lote	Después de procesar un lote lógico

// Punto de control cada 100 eventos
private int _eventCount = 0;

processor.ProcessEventAsync += async args =>
{
    // Procesar evento...
    
    _eventCount++;
    if (_eventCount >= 100)
    {
        await args.UpdateCheckpointAsync();
        _eventCount = 0;
    }
};

SDKs relacionados

SDK	Propósito	Instalación
Azure.Messaging.EventHubs	Envío/recepción principal	dotnet add package Azure.Messaging.EventHubs
Azure.Messaging.EventHubs.Processor	Procesamiento en producción	dotnet add package Azure.Messaging.EventHubs.Processor
Azure.ResourceManager.EventHubs	Plano de administración (crear hubs)	dotnet add package Azure.ResourceManager.EventHubs
Microsoft.Azure.WebJobs.Extensions.EventHubs	Enlace de Azure Functions	dotnet add package Microsoft.Azure.WebJobs.Extensions.EventHubs