Travail avec Apache Kafka en C# (.NET) : bibliothèque Confluent.Kafka

Bibliothèque Confluent.Kafka pour travailler avec Apache Kafka en C# (.NET)

Apache Kafka est une plateforme distribuée de streaming de données, devenue un standard dans la construction d'architectures orientées événements et de microservices. Pour les développeurs C# (.NET), la bibliothèque la plus populaire et la mieux maintenue est Confluent.Kafka — le client officiel de la société Confluent, construit sur la base du client C haute performance librdkafka. Dans cet article, nous allons détailler comment installer la bibliothèque, créer un producteur et un consommateur, et examiner les meilleures pratiques.

Installation de Confluent.Kafka

Pour commencer, ajoutez le paquet à votre projet via NuGet :

dotnet add package Confluent.Kafka

Ou via la console du gestionnaire de paquets :

Install-Package Confluent.Kafka

Configuration de base du producteur (Producer)

Le producteur envoie des messages vers les topics Kafka. Voici un exemple simple :

using Confluent.Kafka;\n\nvar config = new ProducerConfig\n{\n    BootstrapServers = "localhost:9092"\n};\n\nusing (var producer = new ProducerBuilder<string, string>(config).Build())\n{\n    var message = new Message<string, string>\n    {\n        Key = "user_123",\n        Value = "Bonjour, Kafka!"\n    };\n\n    var deliveryReport = await producer.ProduceAsync("my-topic", message);\n    Console.WriteLine($"Message envoyé. Partition : {deliveryReport.Partition}, Offset : {deliveryReport.Offset}");\n}

Points clés :

\n
• BootstrapServers — adresse d'un ou plusieurs brokers Kafka.
\n
• ProduceAsync — envoi asynchrone avec attente d'accusé de réception.
\n
• Key permet de garantir l'ordre des messages pour une même clé.
\n

Producteur avec sérialisation d'objets complexes

Il est souvent nécessaire d'envoyer non pas des chaînes, mais du JSON ou des données binaires. Utilisez un sérialiseur personnalisé ou la bibliothèque Newtonsoft.Json :

using System.Text.Json;\nusing Confluent.Kafka;\n\nvar config = new ProducerConfig { BootstrapServers = "localhost:9092" };\n\nvar producer = new ProducerBuilder<string, Order>(config)\n    .SetValueSerializer(new JsonSerializer<Order>())\n    .Build();\n\nvar order = new Order { Id = 1, Amount = 150.50m, Customer = "Ivan" };\nawait producer.ProduceAsync("orders", new Message<string, Order> { Key = order.Id.ToString(), Value = order });\n\npublic class Order\n{\n    public int Id { get; set; }\n    public decimal Amount { get; set; }\n    public string Customer { get; set; }\n}\n\npublic class JsonSerializer<T> : ISerializer<T>\n{\n    public byte[] Serialize(T data, SerializationContext context) =>\n        JsonSerializer.SerializeToUtf8Bytes(data);\n}

Configuration du consommateur (Consumer)

Le consommateur lit les messages d'un ou plusieurs topics. Exemple d'abonnement et de traitement :

using Confluent.Kafka;\n\nvar config = new ConsumerConfig\n{\n    BootstrapServers = "localhost:9092",\n    GroupId = "my-consumer-group",\n    AutoOffsetReset = AutoOffsetReset.Earliest\n};\n\nusing (var consumer = new ConsumerBuilder<string, string>(config).Build())\n{\n    consumer.Subscribe("my-topic");\n\n    CancellationTokenSource cts = new CancellationTokenSource();\n    Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };\n\n    try\n    {\n        while (true)\n        {\n            var consumeResult = consumer.Consume(cts.Token);\n            Console.WriteLine($"Reçu : Key={consumeResult.Message.Key}, Value={consumeResult.Message.Value}");\n        }\n    }\n    catch (OperationCanceledException)\n    {\n        consumer.Close();\n    }\n}

Paramètres de configuration importants :

\n
• GroupId — identifiant du groupe de consommateurs. Kafka distribue les partitions entre