Работа с Kafka в C# (.NET): библиотека Confluent.Kafka

Работа с Kafka в C# (.NET): библиотека Confluent.Kafka

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

В этой статье мы рассмотрим ключевые возможности Confluent.Kafka, установку, настройку продюсера и консюмера, обработку ошибок, а также приведем практические примеры кода на C#.

Установка и подключение

Библиотека распространяется через NuGet. Для установки выполните команду:

dotnet add package Confluent.Kafka

После установки добавьте using-директивы в ваш код:

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

Базовые понятия

• Producer (Продюсер) — отправляет сообщения в топики Kafka.
• Consumer (Консюмер) — читает сообщения из топиков.
• Topic (Топик) — именованный канал, куда публикуются сообщения.
• Broker (Брокер) — сервер Kafka, хранящий данные.
• Partition (Партиция) — единица параллелизма внутри топика.

Создание продюсера (Producer)

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

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($"Отправлено в partition [{deliveryResult.Partition}] на offset [{deliveryResult.Offset}]");
    }
    catch (ProduceException<string, string> e)
    {
        Console.WriteLine($"Ошибка отправки: {e.Error.Reason}");
    }
}

В этом примере:

• BootstrapServers — адрес кластера Kafka.
• ProduceAsync — асинхронная отправка сообщения.
• Key — опциональный ключ для определения партиции.
• Value — тело сообщения (может быть JSON, строка, бинарные данные).

Создание консюмера (Consumer)

Консюмер подписывается на топики и обрабатывает сообщения. Важный параметр — GroupId, который определяет группу потребителей:

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($"Получено сообщение: Key = {consumeResult.Message.Key}, Value = {consumeResult.Message.Value}");

            // Ручное подтверждение (commit)
            consumer.Commit(consumeResult);
        }
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("Потребление остановлено.");
    }
    finally
    {
        consumer.Close();
    }
}

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

• AutoOffsetReset — с какой позиции начинать чтение (Earliest — с самого начала, Latest — только новые).
• EnableAutoCommit = false — отключаем авто-подтверждение и используем ручной Commit().
• Consume() — блокирующий метод, ожидающий новые сообщения.
• Обработка OperationCanceledException для корректного завершения.

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

В продакшене важно обрабатывать ошибки подключения, таймауты и сбои брокера. Библиотека предоставляет события для логирования:

var producer = new ProducerBuilder<string, string>(config)
    .SetErrorHandler((_, e) => Console.WriteLine($"Ошибка продюсера: {e.Reason}"))
    .Build();

var consumer