microsoft/azure-eventhub-dotnet

Azure.Messaging.EventHubs (.NET)

Hochdurchsatzfähiges Event-Streaming-SDK zum Senden und Empfangen von Ereignissen über Azure Event Hubs.

Installation

# Kernpaket (Senden und einfaches Empfangen)
dotnet add package Azure.Messaging.EventHubs

# Prozessor-Paket (produktives Empfangen mit Checkpointing)
dotnet add package Azure.Messaging.EventHubs.Processor

# Authentifizierung
dotnet add package Azure.Identity

# Für Checkpointing (erforderlich für EventProcessorClient)
dotnet add package Azure.Storage.Blobs

Aktuelle Versionen: Azure.Messaging.EventHubs v5.12.2, Azure.Messaging.EventHubs.Processor v5.12.2

Umgebungsvariablen

EVENTHUB_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net  # Erforderlich: Vollqualifizierter Event Hubs-Namespace
EVENTHUB_NAME=<event-hub-name>  # Erforderlich: Event Hub-Name
BLOB_STORAGE_CONNECTION_STRING=<storage-connection-string>  # Alternative zur Entra ID-Authentifizierung
BLOB_CONTAINER_NAME=<checkpoint-container>  # Erforderlich: Name des Checkpoint-Containers
EVENTHUB_CONNECTION_STRING=Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=...  # Alternative zur Entra ID-Authentifizierung
AZURE_TOKEN_CREDENTIALS=prod  # Nur erforderlich, wenn DefaultAzureCredential in der Produktion verwendet wird

Authentifizierung

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

// Lokale Entwicklung: DefaultAzureCredential. Produktion: setze AZURE_TOKEN_CREDENTIALS=prod oder AZURE_TOKEN_CREDENTIALS=<specific_credential>
var credential = new DefaultAzureCredential(
    DefaultAzureCredential.DefaultEnvironmentVariableName
);
// Oder verwende direkt ein spezifisches Anmeldeinformation in der Produktion:
// Siehe 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);

Erforderliche RBAC-Rollen:

• Senden: Azure Event Hubs Data Sender
• Empfangen: Azure Event Hubs Data Receiver
• Beides: Azure Event Hubs Data Owner

Client-Typen

Client	Zweck	Wann verwenden
EventHubProducerClient	Ereignisse sofort in Batches senden	Echtzeit-Senden, volle Kontrolle über Batchverarbeitung
EventHubBufferedProducerClient	Automatische Batchverarbeitung mit Hintergrundversand	Szenarien mit hohem Volumen und "Feuer und Vergiss"
EventHubConsumerClient	Einfaches Lesen von Ereignissen	Nur für Prototypen, NICHT für die Produktion
EventProcessorClient	Produktive Ereignisverarbeitung	Immer für das Empfangen in der Produktion verwenden

Kern-Workflow

1. Ereignisse senden (Batch)

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

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

// Einen Batch erstellen (respektiert Größenbeschränkungen automatisch)
using EventDataBatch batch = await producer.CreateBatchAsync();

// Ereignisse zum Batch hinzufügen
var events = new[]
{
    new EventData(BinaryData.FromString("{\"id\": 1, \"message\": \"Hallo\"}")),
    new EventData(BinaryData.FromString("{\"id\": 2, \"message\": \"Welt\"}"))
};

foreach (var eventData in events)
{
    if (!batch.TryAdd(eventData))
    {
        // Batch ist voll - senden und einen neuen erstellen
        await producer.SendAsync(batch);
        batch = await producer.CreateBatchAsync();
        
        if (!batch.TryAdd(eventData))
        {
            throw new Exception("Ereignis zu groß für leeren Batch");
        }
    }
}

// Verbleibende Ereignisse senden
if (batch.Count > 0)
{
    await producer.SendAsync(batch);
}

2. Ereignisse senden (Gepuffert - Hohes 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);

// Erfolg/Fehler beim Senden behandeln
producer.SendEventBatchSucceededAsync += args =>
{
    Console.WriteLine($"Batch gesendet: {args.EventBatch.Count} Ereignisse");
    return Task.CompletedTask;
};

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

// Ereignisse in die Warteschlange stellen (werden automatisch im Hintergrund gesendet)
for (int i = 0; i < 1000; i++)
{
    await producer.EnqueueEventAsync(new EventData($"Ereignis {i}"));
}

// Verbleibende Ereignisse vor dem Freigeben leeren
await producer.FlushAsync();

3. Ereignisse empfangen (Produktion - EventProcessorClient)

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

// Blob-Container für Checkpointing
var blobClient = new BlobContainerClient(
    Environment.GetEnvironmentVariable("BLOB_STORAGE_CONNECTION_STRING"),
    Environment.GetEnvironmentVariable("BLOB_CONTAINER_NAME"));

await blobClient.CreateIfNotExistsAsync();

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

// Ereignisse behandeln
processor.ProcessEventAsync += async args =>
{
    Console.WriteLine($"Partition: {args.Partition.PartitionId}");
    Console.WriteLine($"Daten: {args.Data.EventBody}");
    
    // Checkpoint nach der Verarbeitung (oder Batch-Checkpoints)
    await args.UpdateCheckpointAsync();
};

// Fehler behandeln
processor.ProcessErrorAsync += args =>
{
    Console.WriteLine($"Fehler: {args.Exception.Message}");
    Console.WriteLine($"Partition: {args.PartitionId}");
    return Task.CompletedTask;
};

// Verarbeitung starten
await processor.StartProcessingAsync();

// Bis zur Stornierung ausführen
await Task.Delay(Timeout.Infinite, cancellationToken);

// Ordentlich beenden
await processor.StopProcessingAsync();

4. Partitionsoperationen

// Partitions-IDs abrufen
string[] partitionIds = await producer.GetPartitionIdsAsync();

// An eine bestimmte Partition senden (sparsam verwenden)
var options = new SendEventOptions
{
    PartitionId = "0"
};
await producer.SendAsync(events, options);

// Partitionsschlüssel verwenden (empfohlen für Reihenfolge)
var batchOptions = new CreateBatchOptions
{
    PartitionKey = "customer-123"  // Ereignisse mit gleichem Schlüssel kommen in dieselbe Partition
};
using var batch = await producer.CreateBatchAsync(batchOptions);

EventPosition-Optionen

Kontrolliere, wo mit dem Lesen begonnen wird:

// Am Anfang beginnen
EventPosition.Earliest

// Am Ende beginnen (nur neue Ereignisse)
EventPosition.Latest

// Bei einem bestimmten Offset beginnen
EventPosition.FromOffset(12345)

// Bei einer bestimmten Sequenznummer beginnen
EventPosition.FromSequenceNumber(100)

// Zu einem bestimmten Zeitpunkt beginnen
EventPosition.FromEnqueuedTime(DateTimeOffset.UtcNow.AddHours(-1))

ASP.NET Core-Integration

// 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());
});

// In Controller/Service injizieren
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);
    }
}

Best Practices

1. EventProcessorClient für den Empfang verwenden — EventHubConsumerClient niemals in der Produktion nutzen
2. Strategisch checkpointen — Nach N Ereignissen oder einem Zeitintervall, nicht nach jedem Ereignis
3. Partitionsschlüssel verwenden — Für Reihenfolgegarantien innerhalb einer Partition
4. Clients wiederverwenden — Einmal erstellen, als Singleton verwenden (threadsicher)
5. await using verwenden — Stellt ordnungsgemäße Freigabe sicher
6. ProcessErrorAsync behandeln — Immer einen Fehlerhandler registrieren
7. Ereignisse bündeln — CreateBatchAsync() verwenden, um Größenbeschränkungen einzuhalten
8. Gepufferten Producer verwenden — Für Szenarien mit hohem Volumen und automatischer Batchverarbeitung

Fehlerbehandlung

using Azure.Messaging.EventHubs;

try
{
    await producer.SendAsync(batch);
}
catch (EventHubsException ex) when (ex.Reason == EventHubsException.FailureReason.ServiceBusy)
{
    // Wiederholen mit Backoff
    await Task.Delay(TimeSpan.FromSeconds(5));
}
catch (EventHubsException ex) when (ex.IsTransient)
{
    // Transienter Fehler - sicher zu wiederholen
    Console.WriteLine($"Transienter Fehler: {ex.Message}");
}
catch (EventHubsException ex)
{
    // Nicht-transienter Fehler
    Console.WriteLine($"Fehler: {ex.Reason} - {ex.Message}");
}

Checkpointing-Strategien

Strategie	Wann verwenden
Jedes Ereignis	Geringes Volumen, kritische Daten
Alle N Ereignisse	Ausgewogener Durchsatz/Zuverlässigkeit
Zeitbasiert	Konsistente Checkpoint-Intervalle
Batch-Abschluss	Nach der Verarbeitung eines logischen Batches

// Checkpoint alle 100 Ereignisse
private int _eventCount = 0;

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

Verwandte SDKs

SDK	Zweck	Installation
Azure.Messaging.EventHubs	Kernfunktionen Senden/Empfangen	dotnet add package Azure.Messaging.EventHubs
Azure.Messaging.EventHubs.Processor	Produktive Verarbeitung	dotnet add package Azure.Messaging.EventHubs.Processor
Azure.ResourceManager.EventHubs	Verwaltungsebene (Hubs erstellen)	dotnet add package Azure.ResourceManager.EventHubs
Microsoft.Azure.WebJobs.Extensions.EventHubs	Azure Functions-Bindung	dotnet add package Microsoft.Azure.WebJobs.Extensions.EventHubs