Azure Event Hubs bibliothèque cliente du processeur d’événements pour .NET - version 5.9.3

Article
09/12/2023

Azure Event Hubs est un service de publication/abonnement hautement évolutif qui peut ingérer des millions d’événements par seconde et les diffuser en continu vers plusieurs consommateurs. Cela vous permet de traiter et d’analyser les quantités massives de données produites par vos appareils et applications connectés. Une fois qu’Event Hubs a collecté les données, vous pouvez les récupérer, les transformer et les stocker à l’aide de n’importe quel fournisseur d’analytique en temps réel ou avec des adaptateurs de traitement par lots/de stockage. Si vous souhaitez en savoir plus sur Azure Event Hubs, vous pouvez consulter : Qu’est-ce qu’Event Hubs?

La bibliothèque cliente du processeur d’événements est un compagnon de la bibliothèque cliente Azure Event Hubs, en fournissant un client autonome pour la consommation d’événements d’une manière robuste, durable et évolutive qui convient à la plupart des scénarios de production. Implémentation avisée générée à l’aide d’objets blob stockage Azure, le processeur d’événements est recommandé pour :

Lecture et traitement des événements sur toutes les partitions d’un Event Hub à grande échelle avec résilience aux défaillances temporaires et aux problèmes réseau intermittents.
Traitement des événements en coopération, où plusieurs processeurs distribuent et partagent dynamiquement la responsabilité dans le contexte d’un groupe de consommateurs, en gérant correctement la charge à mesure que les processeurs sont ajoutés et supprimés du groupe.
Gestion des points de contrôle et de l’état pour un traitement durable à l’aide d’objets blob stockage Azure comme magasin de données sous-jacent.

| Code source Package (NuGet) | Documentation de référence sur les | API | Documentation produit Guide de résolution des problèmes

Prise en main

Prérequis

Abonnement Azure : Pour utiliser les services Azure, y compris Azure Event Hubs, vous avez besoin d’un abonnement. Si vous n’avez pas de compte Azure existant, vous pouvez vous inscrire à un essai gratuit ou utiliser les avantages de votre abonnement Visual Studio lorsque vous créez un compte.
Espace de noms Event Hubs avec un hub d’événements : Pour interagir avec Azure Event Hubs, vous devez également disposer d’un espace de noms et d’Un hub d’événements. Si vous n’êtes pas familiarisé avec la création de ressources Azure, vous pouvez suivre le guide pas à pas pour créer un Event Hub à l’aide de l’Portail Azure. Vous y trouverez également des instructions détaillées sur l’utilisation des modèles Azure CLI, Azure PowerShell ou Azure Resource Manager (ARM) pour créer un hub d’événements.
Compte de stockage Azure avec stockage d’objets blob : Pour conserver les points de contrôle et régir la propriété dans Stockage Azure, vous devez disposer d’un compte stockage Azure avec des objets blob disponibles. La suppression réversible et le contrôle de version des objets blob doivent être désactivés pour le compte de stockage Azure utilisé pour le processeur. Si vous n’êtes pas familiarisé avec les comptes de stockage Azure, vous pouvez suivre le guide pas à pas pour créer un compte de stockage à l’aide de la Portail Azure. Vous y trouverez également des instructions détaillées sur l’utilisation des modèles Azure CLI, Azure PowerShell ou Azure Resource Manager (ARM) pour créer des comptes de stockage.
Conteneur d’objets blob stockage Azure : Les données de point de contrôle et de propriété dans stockage Azure sont écrites dans des objets blob dans un conteneur spécifique. Nécessite EventProcessorClient un conteneur existant et n’en crée pas implicitement un pour vous protéger contre une configuration incorrecte accidentelle. Il est recommandé d’utiliser un conteneur unique pour chaque combinaison Event Hub et groupe de consommateurs. Si vous n’êtes pas familiarisé avec les conteneurs stockage Azure, vous pouvez consulter la documentation sur la gestion des conteneurs. Vous y trouverez des instructions détaillées sur l’utilisation de .NET, d’Azure CLI ou de Azure PowerShell pour créer un conteneur.
C# 8.0 : La bibliothèque cliente Azure Event Hubs utilise les nouvelles fonctionnalités introduites dans C# 8.0. Pour tirer parti de la syntaxe C# 8.0, il est recommandé de compiler à l’aide du SDK .NET Core 3.0 ou version ultérieure avec une version de langage de latest.

Les utilisateurs de Visual Studio qui souhaitent tirer pleinement parti de la syntaxe C# 8.0 devront utiliser Visual Studio 2019 ou version ultérieure. Visual Studio 2019, y compris l’édition Community gratuite, est téléchargeable ici. Les utilisateurs de Visual Studio 2017 peuvent tirer parti de la syntaxe C# 8 en utilisant le package NuGet Microsoft.Net.Compilers et en définissant la version du langage, bien que l’expérience d’édition ne soit pas idéale.

Vous pouvez toujours utiliser la bibliothèque avec les versions précédentes du langage C#, mais vous devez gérer manuellement les membres asynchrones énumérables et asynchrones jetables au lieu de tirer parti de la nouvelle syntaxe. Vous pouvez toujours cibler n’importe quelle version de framework prise en charge par votre SDK .NET Core, y compris les versions antérieures de .NET Core ou du .NET Framework. Pour plus d’informations, consultez : comment spécifier des frameworks cibles.

Remarque importante : Pour générer ou exécuter les exemples et les exemples sans modification, l’utilisation de C# 11.0 est nécessaire. Vous pouvez toujours exécuter les exemples si vous décidez de les adapter à d’autres versions linguistiques.

Pour créer rapidement les ressources nécessaires dans Azure et recevoir des chaînes de connexion, vous pouvez déployer notre exemple de modèle en cliquant sur :

Installer le package

Installez la bibliothèque cliente du processeur d’événements Azure Event Hubs pour .NET à l’aide de NuGet :

dotnet add package Azure.Messaging.EventHubs.Processor

Authentifier le client

Obtenir une chaîne de connexion Event Hubs

Pour que la bibliothèque cliente Event Hubs interagisse avec un hub d’événements, elle doit comprendre comment se connecter et autoriser avec elle. Le moyen le plus simple consiste à utiliser une chaîne de connexion, qui est créée automatiquement lors de la création d’un espace de noms Event Hubs. Si vous n’êtes pas familiarisé avec l’utilisation de chaînes de connexion avec Event Hubs, vous pouvez suivre le guide pas à pas pour obtenir une chaîne de connexion Event Hubs.

Obtenir une chaîne de connexion stockage Azure

Pour que le client du processeur d’événements utilise des objets blob de stockage Azure pour les points de contrôle, il doit comprendre comment se connecter à un compte de stockage et autoriser avec celui-ci. La méthode la plus simple consiste à utiliser une chaîne de connexion, qui est générée au moment de la création du compte de stockage. Si vous n’êtes pas familiarisé avec l’autorisation de chaîne de connexion de compte de stockage dans Azure, vous pouvez suivre le guide pas à pas pour configurer les chaînes de connexion stockage Azure.

Une fois que vous avez les chaînes de connexion, consultez Création d’un client de processeur d’événements pour obtenir un exemple de leur utilisation pour créer le processeur.

Concepts clés

Un processeur d’événements est une construction destinée à gérer les responsabilités associées à la connexion à un Event Hub donné et au traitement des événements à partir de chacune de ses partitions, dans le contexte d’un groupe de consommateurs spécifique. L’action de traitement des événements lus à partir de la partition et de gestion des erreurs qui se produisent est déléguée par le processeur d’événements au code que vous fournissez, ce qui permet à votre logique de se concentrer sur la fourniture de valeur métier pendant que le processeur gère les tâches associées à la lecture des événements, à la gestion des partitions et à la persistance de l’état sous la forme de points de contrôle.
Le point de contrôle est un processus par lequel les lecteurs marquent et conservent leur position pour les événements qui ont été traités pour une partition. Les points de contrôle sont la responsabilité du consommateur et se produisent sur une partition par partition, généralement dans le contexte d’un groupe de consommateurs spécifique. Pour le EventProcessorClient, cela signifie que, pour un groupe de consommateurs et une combinaison de partitions, le processeur doit effectuer le suivi de sa position actuelle dans le flux d’événements. Si vous souhaitez plus d’informations, reportez-vous aux points de contrôle dans la documentation du produit Event Hubs.

Lorsqu’un processeur d’événements se connecte, il commence à lire les événements au point de contrôle qui a été précédemment conservé par le dernier processeur de cette partition dans ce groupe de consommateurs, le cas échéant. Quand un processeur d’événements lit et agit sur les événements dans la partition, il doit créer régulièrement des points de contrôle pour marquer les événements comme « terminés » par les applications en aval et pour fournir une résilience en cas de défaillance d’un processeur d’événements ou de l’environnement qui l’héberge. Si nécessaire, il est possible de retraiter les événements précédemment marqués comme « terminés » en spécifiant un décalage antérieur dans ce processus de point de contrôle.
Une partition constitue une séquence ordonnée d’événements conservée dans un hub d’événements. Les partitions représentent un moyen d’organiser les données associé au parallélisme requis par les consommateurs d’événements. Azure Event Hubs assure la diffusion de messages suivant un modèle de consommateur partitionné dans lequel chaque consommateur ne lit qu’un sous-ensemble spécifique, ou partition, du flux de message. Les événements les plus récents sont ajoutés à la fin de cette séquence. Le nombre de partitions est spécifié lors de la création du hub d’événements. Il n’est pas modifiable.
Un groupe de consommateurs constitue une vue de tout un hub d’événements. Les groupes de consommateurs permettent à plusieurs applications consommatrices de disposer chacune d’une vue distincte du flux d’événements, et de lire le flux séparément, à son propre rythme et à partir de sa propre position. Il peut y avoir au maximum cinq lecteurs simultanés sur une partition par groupe de consommateurs. Toutefois, il est recommandé de se limiter à un seul consommateur actif pour une association donnée entre une partition et un groupe de consommateurs. Chaque lecteur actif reçoit tous les événements de sa partition. Si plusieurs lecteurs se trouvent sur la même partition, ils reçoivent des événements en double.

Pour plus de concepts et une discussion plus approfondie, consultez Fonctionnalités Event Hubs.

Durée de vie du client

Le EventProcessorClient est sûr à mettre en cache et à utiliser en tant que singleton pendant la durée de vie de l’application, ce qui est une bonne pratique lorsque les événements sont lus régulièrement. Les clients sont responsables de la gestion efficace de l’utilisation du réseau, du processeur et de la mémoire, tout en conservant une utilisation faible pendant les périodes d’inactivité. L’appel StopProcessingAsync ou StopProcessing sur le processeur est nécessaire pour s’assurer que les ressources réseau et autres objets non managés sont correctement nettoyés.

Sécurité des threads

Nous garantissons que toutes les méthodes de instance client sont thread-safe et indépendantes les unes des autres (instructions). Cela garantit que la recommandation de réutilisation des instances clientes est toujours sécurisée, même entre les threads.

Les types de modèle de données, tels que EventData et EventDataBatch ne sont pas thread-safe. Ils ne doivent pas être partagés entre les threads ni utilisés simultanément avec les méthodes clientes.

Concepts supplémentaires

Exemples

Création d’un client de processeur d’événements

Étant donné que EventProcessorClient présente une dépendance vis-à-vis d’objets blob Stockage Azure pour la persistance de son état, vous devez fournir pour le processeur un BlobContainerClient configuré pour le compte de stockage et le conteneur à utiliser. Le conteneur utilisé pour configurer le EventProcessorClient doit exister.

Étant donné que le EventProcessorClient n’a aucun moyen de connaître l’intention de spécifier un conteneur qui n’existe pas, il ne crée pas implicitement le conteneur. Cela sert de protection contre un conteneur mal configuré, ce qui empêche un processeur non autorisé de partager la propriété et d’interférer avec d’autres processeurs du groupe de consommateurs.

// The container specified when creating the BlobContainerClient must exist; it will
// not be implicitly created.

var storageConnectionString = "<< CONNECTION STRING FOR THE STORAGE ACCOUNT >>";
var blobContainerName = "<< NAME OF THE BLOB CONTAINER >>";

var eventHubsConnectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";
var consumerGroup = "<< NAME OF THE EVENT HUB CONSUMER GROUP >>";

var storageClient = new BlobContainerClient(storageConnectionString, blobContainerName);
var processor = new EventProcessorClient(storageClient, consumerGroup, eventHubsConnectionString, eventHubName);

Configurer les gestionnaires d’événements et d’erreurs

Pour pouvoir utiliser , des gestionnaires pour le EventProcessorClienttraitement des événements et les erreurs doivent être fournis. Ces gestionnaires sont considérés comme autonomes et les développeurs sont chargés de s’assurer que les exceptions dans le code du gestionnaire sont prises en compte.

var storageConnectionString = "<< CONNECTION STRING FOR THE STORAGE ACCOUNT >>";
var blobContainerName = "<< NAME OF THE BLOB CONTAINER >>";

var eventHubsConnectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";
var consumerGroup = "<< NAME OF THE EVENT HUB CONSUMER GROUP >>";

async Task processEventHandler(ProcessEventArgs eventArgs)
{
    try
    {
        // Perform the application-specific processing for an event.  This method
        // is intended for illustration and is not defined in this snippet.

        await DoSomethingWithTheEvent(eventArgs.Partition, eventArgs.Data);
    }
    catch
    {
        // Handle the exception from handler code
    }
}

async Task processErrorHandler(ProcessErrorEventArgs eventArgs)
{
    try
    {
        // Perform the application-specific processing for an error.  This method
        // is intended for illustration and is not defined in this snippet.

        await DoSomethingWithTheError(eventArgs.Exception);
    }
    catch
    {
        // Handle the exception from handler code
    }
}

var storageClient = new BlobContainerClient(storageConnectionString, blobContainerName);
var processor = new EventProcessorClient(storageClient, consumerGroup, eventHubsConnectionString, eventHubName);

processor.ProcessEventAsync += processEventHandler;
processor.ProcessErrorAsync += processErrorHandler;

Démarrer et arrêter le traitement

Le EventProcessorClient effectue son traitement en arrière-plan une fois qu’il a été explicitement démarré et continue de le faire jusqu’à ce qu’il ait été explicitement arrêté. Bien que cela permette au code de l’application d’effectuer d’autres tâches, il lui incombe également de s’assurer que le processus ne s’arrête pas pendant le traitement s’il n’y a pas d’autres tâches en cours d’exécution.

var cancellationSource = new CancellationTokenSource();
cancellationSource.CancelAfter(TimeSpan.FromSeconds(45));

var storageConnectionString = "<< CONNECTION STRING FOR THE STORAGE ACCOUNT >>";
var blobContainerName = "<< NAME OF THE BLOB CONTAINER >>";

var eventHubsConnectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";
var consumerGroup = "<< NAME OF THE EVENT HUB CONSUMER GROUP >>";

Task processEventHandler(ProcessEventArgs eventArgs) => Task.CompletedTask;
Task processErrorHandler(ProcessErrorEventArgs eventArgs) => Task.CompletedTask;

var storageClient = new BlobContainerClient(storageConnectionString, blobContainerName);
var processor = new EventProcessorClient(storageClient, consumerGroup, eventHubsConnectionString, eventHubName);

processor.ProcessEventAsync += processEventHandler;
processor.ProcessErrorAsync += processErrorHandler;

await processor.StartProcessingAsync();

try
{
    // The processor performs its work in the background; block until cancellation
    // to allow processing to take place.

    await Task.Delay(Timeout.Infinite, cancellationSource.Token);
}
catch (TaskCanceledException)
{
    // This is expected when the delay is canceled.
}

try
{
    await processor.StopProcessingAsync();
}
finally
{
    // To prevent leaks, the handlers should be removed when processing is complete.

    processor.ProcessEventAsync -= processEventHandler;
    processor.ProcessErrorAsync -= processErrorHandler;
}

Utilisation d’un principal Active Directory avec le client du processeur d’événements

La bibliothèque Azure Identity fournit une prise en charge de l’authentification Azure Active Directory qui peut être utilisée pour les bibliothèques clientes Azure, notamment Event Hubs et Stockage Azure.

Pour utiliser un principal Active Directory, l’une des informations d’identification disponibles à partir de la Azure.Identity bibliothèque est spécifiée lors de la création du client Event Hubs. En outre, l’espace de noms Event Hubs complet et le nom du hub d’événements souhaité sont fournis à la place de la chaîne de connexion Event Hubs.

Pour utiliser un principal Active Directory avec des conteneurs d’objets blob stockage Azure, l’URL complète du conteneur doit être fournie lors de la création du client de stockage. Pour plus d’informations sur les formats d’URI valides pour accéder au stockage d’objets blob , consultez Nommage et référencement des conteneurs, objets blob et métadonnées.

var credential = new DefaultAzureCredential();
var blobStorageUrl ="<< FULLY-QUALIFIED CONTAINER URL (like https://myaccount.blob.core.windows.net/mycontainer) >>";

var fullyQualifiedNamespace = "<< FULLY-QUALIFIED EVENT HUBS NAMESPACE (like something.servicebus.windows.net) >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";
var consumerGroup = "<< NAME OF THE EVENT HUB CONSUMER GROUP >>";

var storageClient = new BlobContainerClient(new Uri(blobStorageUrl), credential);

var processor = new EventProcessorClient
(
    storageClient,
    consumerGroup,
    fullyQualifiedNamespace,
    eventHubName,
    credential
);

Lorsque vous utilisez Azure Active Directory avec Event Hubs, un rôle qui permet la lecture à partir d’Event Hubs doit être attribué à votre principal, tel que le Azure Event Hubs Data Receiver rôle. Pour plus d’informations sur l’utilisation de l’autorisation Azure Active Directory avec Event Hubs, consultez la documentation associée.

Lorsque vous utilisez Azure Active Directory avec Stockage Azure, un rôle doit être attribué à votre principal qui autorise l’accès en lecture, écriture et suppression aux objets blob, tels que le Storage Blob Data Contributor rôle. Pour plus d’informations sur l’utilisation de l’autorisation Active Directory avec Stockage Azure, reportez-vous à la documentation associée et à l’exemple d’autorisation stockage Azure.

Dépannage

Pour obtenir des informations détaillées sur la résolution des problèmes, reportez-vous au Guide de résolution des problèmes d’Event Hubs.

Gestion des exceptions

Exceptions clientes du processeur d’événements

Le client Event Processor fait toutes les tentatives pour être résilient face aux exceptions et prendra les mesures nécessaires pour poursuivre le traitement, sauf s’il est impossible de le faire. Aucune action des développeurs n’est nécessaire pour que cela se produise ; il fait partie nativement du comportement du processeur.

Afin de permettre aux développeurs d’inspecter et de réagir aux exceptions qui se produisent dans les opérations clientes du processeur d’événements, elles sont exposées via l’événement ProcessError . Les arguments de cet événement fournissent des détails sur l’exception et le contexte dans lequel elle a été observée. Les développeurs peuvent effectuer des opérations normales sur le client Event Processor à partir de ce gestionnaire d’événements, par exemple l’arrêter et/ou le redémarrer en réponse à des erreurs, mais ils ne peuvent pas influencer le comportement des exceptions du processeur.

Pour obtenir un exemple de base d’implémentation du gestionnaire d’erreurs, consultez l’exemple : Gestionnaires du processeur d’événements.

Exceptions dans les gestionnaires d’événements

Étant donné que le client Du processeur d’événements ne dispose pas du contexte approprié pour comprendre la gravité des exceptions dans les gestionnaires d’événements fournis par les développeurs, il ne peut pas supposer quelles actions constitueraient une réponse raisonnable. Par conséquent, les développeurs sont considérés comme responsables des exceptions qui se produisent dans les gestionnaires d’événements qu’ils fournissent à l’aide try/catch de blocs et d’autres constructions de langage standard.

Le client Event Processor ne tente pas de détecter les exceptions dans le code du développeur ni de les exposer explicitement. Le comportement résultant dépend de l’environnement d’hébergement du processeur et du contexte dans lequel le gestionnaire d’événements a été appelé. Étant donné que cela peut varier d’un scénario à l’autre, il est vivement recommandé aux développeurs de coder leurs gestionnaires d’événements de manière défensive et de tenir compte des exceptions potentielles.

Journalisation et diagnostics

La bibliothèque cliente du processeur d’événements est entièrement instrumentée pour la journalisation des informations à différents niveaux de détail à l’aide du .NET EventSource pour émettre des informations. La journalisation est effectuée pour chaque opération et suit le modèle de marquage du point de départ de l’opération, de son achèvement et de toutes les exceptions rencontrées. Des informations supplémentaires susceptibles d’offrir des insights sont également enregistrées dans le contexte de l’opération associée.

Les journaux du client Event Processor sont disponibles pour n’importe qui EventListener en choisissant la source nommée « Azure-Messaging-EventHubs-Processor-EventProcessorClient » ou en optant pour toutes les sources qui ont la caractéristique « AzureEventSource ». Pour faciliter la capture des journaux à partir des bibliothèques clientes Azure, la Azure.Core bibliothèque utilisée par Event Hubs offre un AzureEventSourceListener. Pour plus d’informations, consultez Capture des journaux Event Hubs à l’aide d’AzureEventSourceListener.

La bibliothèque du processeur d’événements est également instrumentée pour le suivi distribué à l’aide d’Application Insights ou d’OpenTelemetry. Pour plus d’informations, consultez l’exemple Diagnostics Azure.Core.

Étapes suivantes

Au-delà des scénarios abordés, la bibliothèque processeur Azure Event Hubs offre une prise en charge de scénarios supplémentaires pour vous aider à tirer parti de l’ensemble de fonctionnalités complet de .EventProcessorClient Pour vous aider à explorer certains de ces scénarios, la bibliothèque cliente du processeur Event Hubs propose un projet d’exemples pour servir d’illustration pour des scénarios courants. Pour plus d’informations, consultez les exemples README .

Contribution

Ce projet accepte les contributions et les suggestions. La plupart des contributions vous demandent d’accepter un contrat de licence de contribution (CLA) déclarant que vous avez le droit de nous accorder, et que vous nous accordez réellement, les droits d’utilisation de votre contribution. Pour plus d’informations, visitez https://cla.microsoft.com.

Quand vous envoyez une demande de tirage (pull request), un bot CLA détermine automatiquement si vous devez fournir un contrat CLA et agrémenter la demande de tirage de façon appropriée (par exemple, avec une étiquette ou un commentaire). Suivez simplement les instructions fournies par le bot. Vous ne devez effectuer cette opération qu’une seule fois sur tous les dépôts utilisant notre contrat CLA.

Ce projet a adopté le Code de conduite Open Source de Microsoft. Pour plus d’informations, consultez les Questions fréquentes (FAQ) sur le code de conduite ou envoyez vos questions ou vos commentaires à opencode@microsoft.com.

Pour plus d’informations, consultez notre guide de contribution .

Impressions