Arbeiten mit Apache Kafka in C# (.NET): Die Bibliothek Confluent.Kafka

Die Bibliothek Confluent.Kafka für die Arbeit mit Apache Kafka in C# (.NET)

Apache Kafka ist eine verteilte Plattform für die Datenstromverarbeitung, die zum Standard für den Aufbau ereignisgesteuerter Architekturen und Microservices geworden ist. Für Entwickler in C# (.NET) ist die beliebteste und am besten unterstützte Bibliothek Confluent.Kafka – der offizielle Client von Confluent, der auf dem leistungsstarken C-Client librdkafka basiert. In diesem Artikel werden wir im Detail besprechen, wie man die Bibliothek installiert, einen Producer und Consumer erstellt und die besten Praktiken anwendet.

Installation von Confluent.Kafka

Fügen Sie zu Beginn das Paket über NuGet zu Ihrem Projekt hinzu:

dotnet add package Confluent.Kafka

Oder über die Paket-Manager-Konsole:

Install-Package Confluent.Kafka

Grundlegende Konfiguration des Producers

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

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 = "Hallo, Kafka!"\n    };\n\n    var deliveryReport = await producer.ProduceAsync("my-topic", message);\n    Console.WriteLine($"Nachricht gesendet. Partition: {deliveryReport.Partition}, Offset: {deliveryReport.Offset}");\n}

Wichtige Punkte:

\n
• BootstrapServers – Adresse eines oder mehrerer Kafka-Broker.
\n
• ProduceAsync – asynchrones Senden mit Warten auf Bestätigung.
\n
• Der Key ermöglicht die Sicherstellung der Reihenfolge von Nachrichten mit demselben Schlüssel.
\n

Producer mit Serialisierung komplexer Objekte

Oft müssen nicht nur Zeichenfolgen, sondern JSON oder Binärdaten gesendet werden. Verwenden Sie einen benutzerdefinierten Serialisierer oder die Bibliothek 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}

Konfiguration des Consumers

Der Consumer liest Nachrichten aus einem oder mehreren Topics. Beispiel für das Abonnieren und Verarbeiten:

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($"Empfangen: Key={consumeResult.Message.Key}, Value={consumeResult.Message.Value}");\n        }\n    }\n    catch (OperationCanceledException)\n    {\n        consumer.Close();\n    }\n}

Wichtige Konfigurationsparameter:

\n
• GroupId – Identifikator der Consumergruppe. Kafka verteilt die Partitionen zwischen