Arbeit mit Kafka in C# (.NET): Bibliothek Confluent.Kafka

Arbeit mit Kafka in C# (.NET): Bibliothek Confluent.Kafka

Apache Kafka ist eine verteilte Plattform für die Datenstromverarbeitung, die häufig für den Aufbau hochbelastbarer Nachrichtensysteme, Ereignisverarbeitung und Microservice-Integration verwendet wird. Für die Arbeit mit Kafka im .NET-Ökosystem ist die wichtigste und am besten unterstützte Bibliothek Confluent.Kafka – der offizielle Client von Confluent, der auf der leistungsstarken C-Bibliothek librdkafka basiert.

In diesem Artikel betrachten wir die wichtigsten Funktionen von Confluent.Kafka, Installation, Konfiguration von Producer und Consumer, Fehlerbehandlung und geben praktische Codebeispiele in C#.

Installation und Einbindung

Die Bibliothek wird über NuGet vertrieben. Führen Sie zur Installation folgenden Befehl aus:

dotnet add package Confluent.Kafka

Fügen Sie nach der Installation die using-Direktiven in Ihren Code ein:

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

Grundlegende Konzepte

• Producer – sendet Nachrichten an Kafka-Topics.
• Consumer – liest Nachrichten aus Topics.
• Topic – ein benannter Kanal, in den Nachrichten veröffentlicht werden.
• Broker – Kafka-Server, der die Daten speichert.
• Partition – Einheit der Parallelität innerhalb eines Topics.

Erstellen eines Producers

Der Producer sendet Nachrichten an Kafka. Betrachten wir ein einfaches Beispiel:

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 = "Hello Kafka from .NET!" }        );

        Console.WriteLine($"Gesendet an Partition [{deliveryResult.Partition}] auf Offset [{deliveryResult.Offset}]");    }    catch (ProduceException<string, string> e)    {        Console.WriteLine($"Fehler beim Senden: {e.Error.Reason}");    }}

In diesem Beispiel:

• BootstrapServers – Adresse des Kafka-Clusters.
• ProduceAsync – asynchrones Senden der Nachricht.
• Key – optionaler Schlüssel zur Bestimmung der Partition.
• Value – Nachrichtentext (kann JSON, String, Binärdaten sein).

Erstellen eines Consumers

Der Consumer abonniert Topics und verarbeitet Nachrichten. Ein wichtiger Parameter ist GroupId, der die Verbrauchergruppe definiert:

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($"Nachricht empfangen: Key = {consumeResult.Message.Key}, Value = {consumeResult.Message.Value}");

            // Manuelle Bestätigung (Commit)            consumer.Commit(consumeResult);        }    }    catch (OperationCanceledException)    {        Console.WriteLine("Konsum gestoppt.");    }    finally    {        consumer.Close();    }}

Wichtige Punkte:

• AutoOff