Travailler avec Kafka en C# (.NET) : la bibliothèque Confluent.Kafka

Travailler avec Kafka en C# (.NET) : la bibliothèque Confluent.Kafka

Apache Kafka est une plateforme distribuée de streaming de données, largement utilisée pour construire des systèmes de messagerie à haute charge, le traitement d'événements et l'intégration de microservices. Pour travailler avec Kafka dans l'écosystème .NET, la bibliothèque principale et la mieux prise en charge est Confluent.Kafka — le client officiel de Confluent, construit sur la bibliothèque C haute performance librdkafka.

Dans cet article, nous examinerons les fonctionnalités clés de Confluent.Kafka, l'installation, la configuration du producteur et du consommateur, la gestion des erreurs, et nous fournirons des exemples pratiques de code en C#.

Installation et configuration

La bibliothèque est distribuée via NuGet. Pour l'installer, exécutez la commande :

dotnet add package Confluent.Kafka

Après l'installation, ajoutez les directives using dans votre code :

using Confluent.Kafka;using Confluent.Kafka.Admin;using System;using System.Threading;using System.Threading.Tasks;

Concepts de base

• Producer (Producteur) — envoie des messages vers les topics Kafka.
• Consumer (Consommateur) — lit les messages depuis les topics.
• Topic (Sujet) — canal nommé où les messages sont publiés.
• Broker (Courtier) — serveur Kafka qui stocke les données.
• Partition (Partition) — unité de parallélisme au sein d'un topic.

Création d'un producteur (Producer)

Le producteur envoie des messages vers Kafka. Voici un exemple de base :

var config = new ProducerConfig{    BootstrapServers = "localhost:9092"};

using (var producer = new ProducerBuilder<string, string>(config).Build()){    try    {        var deliveryResult = await producer.ProduceAsync(            "my-topic",            new Message<string, string> { Key = "key1", Value = "Bonjour Kafka depuis .NET!" }        );

        Console.WriteLine($"Envoyé vers la partition [{deliveryResult.Partition}] à l'offset [{deliveryResult.Offset}]");    }    catch (ProduceException<string, string> e)    {        Console.WriteLine($"Erreur d'envoi : {e.Error.Reason}");    }}

Dans cet exemple :

• BootstrapServers — adresse du cluster Kafka.
• ProduceAsync — envoi asynchrone du message.
• Key — clé optionnelle pour déterminer la partition.
• Value — corps du message (peut être JSON, chaîne, données binaires).

Création d'un consommateur (Consumer)

Le consommateur s'abonne aux topics et traite les messages. Un paramètre important est GroupId, qui définit le groupe de consommateurs :

var config = new ConsumerConfig{    BootstrapServers = "localhost:9092",    GroupId = "my-consumer-group",    AutoOffsetReset = AutoOffsetReset.Earliest,    EnableAutoCommit = false};

using (var consumer = new ConsumerBuilder<string, string>(config).Build()){    consumer.Subscribe("my-topic");

    CancellationTokenSource cts = new CancellationTokenSource();    Console.CancelKeyPress += (_, e) => {        e.Cancel = true;        cts.Cancel();    };

    try    {        while (true)        {            var consumeResult = consumer.Consume(cts.Token);

            Console.WriteLine($"Message reçu : Key = {consumeResult.Message.Key}, Value = {consumeResult.Message.Value}");

            // Confirmation manuelle (commit)            consumer.Commit(consumeResult);        }    }    catch (OperationCanceledException)    {        Console.WriteLine("Consommation arrêtée.");    }    finally    {        consumer.Close();    }}

Points clés :

• AutoOff