microsoft/azure-eventhub-dotnet

Azure.Messaging.EventHubs (.NET)

SDK de streaming d'événements à haut débit pour envoyer et recevoir des événements via Azure Event Hubs.

Installation

# Package principal (envoi et réception simple)
dotnet add package Azure.Messaging.EventHubs

# Package processeur (réception en production avec checkpointing)
dotnet add package Azure.Messaging.EventHubs.Processor

# Authentification
dotnet add package Azure.Identity

# Pour le checkpointing (requis par EventProcessorClient)
dotnet add package Azure.Storage.Blobs

Versions actuelles : Azure.Messaging.EventHubs v5.12.2, Azure.Messaging.EventHubs.Processor v5.12.2

Variables d'environnement

EVENTHUB_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net  # Requis : espace de noms qualifié complet Event Hubs
EVENTHUB_NAME=<event-hub-name>  # Requis : nom de l'Event Hub
BLOB_STORAGE_CONNECTION_STRING=<storage-connection-string>  # Alternative à l'authentification Entra ID
BLOB_CONTAINER_NAME=<checkpoint-container>  # Requis : nom du conteneur de checkpoint
EVENTHUB_CONNECTION_STRING=Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=...  # Alternative à l'authentification Entra ID
AZURE_TOKEN_CREDENTIALS=prod  # Requis uniquement si DefaultAzureCredential est utilisé en production

Authentification

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

// Développement local : DefaultAzureCredential. Production : définir AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=<credential_spécifique>
var credential = new DefaultAzureCredential(
    DefaultAzureCredential.DefaultEnvironmentVariableName
);
// Ou utiliser directement une information d'identification spécifique en production :
// Voir 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);

Rôles RBAC requis :

• Envoi : Azure Event Hubs Data Sender
• Réception : Azure Event Hubs Data Receiver
• Les deux : Azure Event Hubs Data Owner

Types de clients

Client	Objectif	Quand l'utiliser
EventHubProducerClient	Envoyer des événements immédiatement par lots	Envoi en temps réel, contrôle total du regroupement
EventHubBufferedProducerClient	Regroupement automatique avec envoi en arrière-plan	Scénarios à volume élevé, "fire-and-forget"
EventHubConsumerClient	Lecture d'événements simple	Prototypage uniquement, PAS pour la production
EventProcessorClient	Traitement d'événements en production	Toujours utiliser celui-ci pour la réception en production

Flux de travail principal

1. Envoyer des événements (par lots)

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

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

// Créer un lot (respecte automatiquement les limites de taille)
using EventDataBatch batch = await producer.CreateBatchAsync();

// Ajouter des événements au lot
var events = new[]
{
    new EventData(BinaryData.FromString("{\"id\": 1, \"message\": \"Bonjour\"}")),
    new EventData(BinaryData.FromString("{\"id\": 2, \"message\": \"Monde\"}"))
};

foreach (var eventData in events)
{
    if (!batch.TryAdd(eventData))
    {
        // Le lot est plein - l'envoyer et en créer un nouveau
        await producer.SendAsync(batch);
        batch = await producer.CreateBatchAsync();
        
        if (!batch.TryAdd(eventData))
        {
            throw new Exception("Événement trop grand pour un lot vide");
        }
    }
}

// Envoyer les événements restants
if (batch.Count > 0)
{
    await producer.SendAsync(batch);
}

2. Envoyer des événements (tamponné - volume élevé)

using Azure.Messaging.EventHubs.Producer;

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

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

// Gérer le succès/échec de l'envoi
producer.SendEventBatchSucceededAsync += args =>
{
    Console.WriteLine($"Lot envoyé : {args.EventBatch.Count} événements");
    return Task.CompletedTask;
};

producer.SendEventBatchFailedAsync += args =>
{
    Console.WriteLine($"Échec du lot : {args.Exception.Message}");
    return Task.CompletedTask;
};

// Mettre les événements en file d'attente (envoyés automatiquement en arrière-plan)
for (int i = 0; i < 1000; i++)
{
    await producer.EnqueueEventAsync(new EventData($"Événement {i}"));
}

// Vider les événements restants avant la suppression
await producer.FlushAsync();

3. Recevoir des événements (Production - EventProcessorClient)

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

// Conteneur Blob pour le checkpointing
var blobClient = new BlobContainerClient(
    Environment.GetEnvironmentVariable("BLOB_STORAGE_CONNECTION_STRING"),
    Environment.GetEnvironmentVariable("BLOB_CONTAINER_NAME"));

await blobClient.CreateIfNotExistsAsync();

// Créer le processeur
var processor = new EventProcessorClient(
    blobClient,
    EventHubConsumerClient.DefaultConsumerGroup,
    fullyQualifiedNamespace,
    eventHubName,
    new DefaultAzureCredential());

// Gérer les événements
processor.ProcessEventAsync += async args =>
{
    Console.WriteLine($"Partition : {args.Partition.PartitionId}");
    Console.WriteLine($"Données : {args.Data.EventBody}");
    
    // Effectuer le checkpoint après le traitement (ou par lots de checkpoints)
    await args.UpdateCheckpointAsync();
};

// Gérer les erreurs
processor.ProcessErrorAsync += args =>
{
    Console.WriteLine($"Erreur : {args.Exception.Message}");
    Console.WriteLine($"Partition : {args.PartitionId}");
    return Task.CompletedTask;
};

// Démarrer le traitement
await processor.StartProcessingAsync();

// Exécuter jusqu'à annulation
await Task.Delay(Timeout.Infinite, cancellationToken);

// Arrêter proprement
await processor.StopProcessingAsync();

4. Opérations sur les partitions

// Obtenir les ID de partition
string[] partitionIds = await producer.GetPartitionIdsAsync();

// Envoyer vers une partition spécifique (à utiliser avec parcimonie)
var options = new SendEventOptions
{
    PartitionId = "0"
};
await producer.SendAsync(events, options);

// Utiliser une clé de partition (recommandé pour l'ordre)
var batchOptions = new CreateBatchOptions
{
    PartitionKey = "client-123"  // Les événements avec la même clé vont dans la même partition
};
using var batch = await producer.CreateBatchAsync(batchOptions);

Options EventPosition

Contrôler où commencer la lecture :

// Commencer depuis le début
EventPosition.Earliest

// Commencer depuis la fin (nouveaux événements uniquement)
EventPosition.Latest

// Commencer depuis un décalage spécifique
EventPosition.FromOffset(12345)

// Commencer depuis un numéro de séquence spécifique
EventPosition.FromSequenceNumber(100)

// Commencer depuis une heure spécifique
EventPosition.FromEnqueuedTime(DateTimeOffset.UtcNow.AddHours(-1))

Intégration 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());
});

// Injection dans le contrôleur/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();
        batch.TryAdd(new EventData(message));
        await _producer.SendAsync(batch);
    }
}

Bonnes pratiques

1. Utiliser EventProcessorClient pour la réception — Ne jamais utiliser EventHubConsumerClient en production
2. Effectuer le checkpoint stratégiquement — Après N événements ou un intervalle de temps, pas après chaque événement
3. Utiliser les clés de partition — Pour des garanties d'ordre au sein d'une partition
4. Réutiliser les clients — Créer une fois, utiliser comme singleton (thread-safe)
5. Utiliser await using — Garantit une suppression correcte
6. Gérer ProcessErrorAsync — Toujours enregistrer un gestionnaire d'erreur
7. Regrouper les événements — Utiliser CreateBatchAsync() pour respecter les limites de taille
8. Utiliser le producteur tamponné — Pour les scénarios à volume élevé avec regroupement automatique

Gestion des erreurs

using Azure.Messaging.EventHubs;

try
{
    await producer.SendAsync(batch);
}
catch (EventHubsException ex) when (ex.Reason == EventHubsException.FailureReason.ServiceBusy)
{
    // Réessayer avec un délai d'attente
    await Task.Delay(TimeSpan.FromSeconds(5));
}
catch (EventHubsException ex) when (ex.IsTransient)
{
    // Erreur transitoire - sans risque de réessayer
    Console.WriteLine($"Erreur transitoire : {ex.Message}");
}
catch (EventHubsException ex)
{
    // Erreur non transitoire
    Console.WriteLine($"Erreur : {ex.Reason} - {ex.Message}");
}

Stratégies de checkpointing

Stratégie	Quand l'utiliser
Chaque événement	Faible volume, données critiques
Tous les N événements	Trait équilibré débit/fiabilité
Basée sur le temps	Intervalles de checkpoint cohérents
À la fin d'un lot	Après le traitement d'un lot logique

// Checkpoint tous les 100 événements
private int _eventCount = 0;

processor.ProcessEventAsync += async args =>
{
    // Traiter l'événement...
    
    _eventCount++;
    if (_eventCount >= 100)
    {
        await args.UpdateCheckpointAsync();
        _eventCount = 0;
    }
};

Kits SDK associés

SDK	Objectif	Installation
Azure.Messaging.EventHubs	Envoi/réception principal	dotnet add package Azure.Messaging.EventHubs
Azure.Messaging.EventHubs.Processor	Traitement en production	dotnet add package Azure.Messaging.EventHubs.Processor
Azure.ResourceManager.EventHubs	Plan de gestion (créer des hubs)	dotnet add package Azure.ResourceManager.EventHubs
Microsoft.Azure.WebJobs.Extensions.EventHubs	Liaison Azure Functions	dotnet add package Microsoft.Azure.WebJobs.Extensions.EventHubs