Работа с Kafka в C# (.NET) | Confluent.Kafka

Библиотека Confluent.Kafka для работы с Apache Kafka в C# (.NET)

Apache Kafka — это распределенная платформа для потоковой передачи данных, ставшая стандартом в построении событийно-ориентированных архитектур и микросервисов. Для разработчиков на C# (.NET) наиболее популярной и поддерживаемой библиотекой является Confluent.Kafka — официальный клиент от компании Confluent, построенный на основе высокопроизводительного C-клиента librdkafka. В этой статье мы подробно разберем, как установить библиотеку, создать продюсера и консюмера, а также рассмотрим лучшие практики работы.

Установка Confluent.Kafka

Для начала работы добавьте пакет в проект через NuGet:

dotnet add package Confluent.Kafka

Или через консоль диспетчера пакетов:

Install-Package Confluent.Kafka

Базовая настройка продюсера (Producer)

Продюсер отправляет сообщения в топики Kafka. Рассмотрим простой пример:

using Confluent.Kafka;

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

using (var producer = new ProducerBuilder<string, string>(config).Build())
{
    var message = new Message<string, string>
    {
        Key = "user_123",
        Value = "Привет, Kafka!"
    };

    var deliveryReport = await producer.ProduceAsync("my-topic", message);
    Console.WriteLine($"Сообщение отправлено. Partition: {deliveryReport.Partition}, Offset: {deliveryReport.Offset}");
}

Ключевые моменты:

BootstrapServers — адрес одного или нескольких брокеров Kafka.
ProduceAsync — асинхронная отправка с ожиданием подтверждения.
Key позволяет гарантировать упорядоченность сообщений по одному ключу.

Продюсер с сериализацией сложных объектов

Часто требуется отправлять не строки, а JSON или бинарные данные. Используйте пользовательский сериализатор или библиотеку Newtonsoft.Json:

using System.Text.Json;
using Confluent.Kafka;

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

var producer = new ProducerBuilder<string, Order>(config)
    .SetValueSerializer(new JsonSerializer<Order>())
    .Build();

var order = new Order { Id = 1, Amount = 150.50m, Customer = "Иван" };
await producer.ProduceAsync("orders", new Message<string, Order> { Key = order.Id.ToString(), Value = order });

public class Order
{
    public int Id { get; set; }
    public decimal Amount { get; set; }
    public string Customer { get; set; }
}

public class JsonSerializer<T> : ISerializer<T>
{
    public byte[] Serialize(T data, SerializationContext context) =>
        JsonSerializer.SerializeToUtf8Bytes(data);
}

Настройка консюмера (Consumer)

Консюмер читает сообщения из одного или нескольких топиков. Пример подписки и обработки:

using Confluent.Kafka;

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

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($"Получено: Key={consumeResult.Message.Key}, Value={consumeResult.Message.Value}");
        }
    }
    catch (OperationCanceledException)
    {
        consumer.Close();
    }
}

Важные параметры конфигурации:

GroupId — идентификатор группы консюмеров. Kafka распределяет партиции между участниками группы.
AutoOffsetReset — определяет, с какого момента начинать чтение, если нет сохраненного смещения. Earliest — с самого начала, Latest — только новые сообщения.
Consume() — блокирующий метод, ожидающий новое сообщение.

Обработка ошибок и повторные попытки

В production-среде необходимо корректно обрабатывать сбои. Confluent.Kafka предоставляет события для логирования и ошибок:

var producer = new ProducerBuilder<string, string>(config)
    .SetErrorHandler((_, error) =>
    {
        Console.WriteLine($"Ошибка: {error.Reason}");
    })
    .SetLogHandler((_, logMessage) =>
    {
        Console.WriteLine($"Лог: {logMessage.Message}");
    })
    .Build();

Для консю