Trabajo con Kafka en C# (.NET): biblioteca Confluent.Kafka

Trabajo con Kafka en C# (.NET): biblioteca Confluent.Kafka

Apache Kafka es una plataforma distribuida para transmisión de datos en streaming, ampliamente utilizada para construir sistemas de mensajería de alta carga, procesamiento de eventos e integración de microservicios. Para trabajar con Kafka en el ecosistema .NET, la biblioteca principal y más compatible es Confluent.Kafka, el cliente oficial de Confluent, construido sobre la biblioteca C de alto rendimiento librdkafka.

En este artículo, exploraremos las capacidades clave de Confluent.Kafka, instalación, configuración del productor y consumidor, manejo de errores y proporcionaremos ejemplos prácticos de código en C#.

Instalación y configuración

La biblioteca se distribuye a través de NuGet. Para instalarla, ejecute el comando:

dotnet add package Confluent.Kafka

Después de la instalación, agregue las directivas using a su código:

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

Conceptos básicos

• Producer (Productor) — envía mensajes a los tópicos de Kafka.
• Consumer (Consumidor) — lee mensajes de los tópicos.
• Topic (Tópico) — canal con nombre donde se publican los mensajes.
• Broker (Broker) — servidor Kafka que almacena datos.
• Partition (Partición) — unidad de paralelismo dentro de un tópico.

Creación de un productor (Producer)

El productor envía mensajes a Kafka. Veamos un ejemplo básico:

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 = "¡Hola Kafka desde .NET!" }        );

        Console.WriteLine($"Enviado a la partición [{deliveryResult.Partition}] en offset [{deliveryResult.Offset}]");    }    catch (ProduceException<string, string> e)    {        Console.WriteLine($"Error de envío: {e.Error.Reason}");    }}

En este ejemplo:

• BootstrapServers — dirección del clúster de Kafka.
• ProduceAsync — envío asíncrono del mensaje.
• Key — clave opcional para determinar la partición.
• Value — cuerpo del mensaje (puede ser JSON, cadena, datos binarios).

Creación de un consumidor (Consumer)

El consumidor se suscribe a los tópicos y procesa los mensajes. Un parámetro importante es GroupId, que define el grupo de consumidores:

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

            // Confirmación manual (commit)            consumer.Commit(consumeResult);        }    }    catch (OperationCanceledException)    {        Console.WriteLine("Consumo detenido.");    }    finally    {        consumer.Close();    }}

Puntos clave:

• AutoOff