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

Biblioteca Confluent.Kafka para trabajar con Apache Kafka en C# (.NET)

Apache Kafka es una plataforma distribuida para transmisión de datos en tiempo real que se ha convertido en el estándar para construir arquitecturas orientadas a eventos y microservicios. Para desarrolladores en C# (.NET), la biblioteca más popular y con soporte es Confluent.Kafka, el cliente oficial de la empresa Confluent, construido sobre el cliente C de alto rendimiento librdkafka. En este artículo analizaremos en detalle cómo instalar la biblioteca, crear un productor y un consumidor, y revisaremos las mejores prácticas de trabajo.

Instalación de Confluent.Kafka

Para comenzar, agregue el paquete al proyecto a través de NuGet:

dotnet add package Confluent.Kafka

O a través de la consola del administrador de paquetes:

Install-Package Confluent.Kafka

Configuración básica del productor (Producer)

El productor envía mensajes a los tópicos de Kafka. Veamos un ejemplo simple:

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

Puntos clave:

\n
• BootstrapServers — dirección de uno o varios brokers de Kafka.
\n
• ProduceAsync — envío asíncrono con espera de confirmación.
\n
• Key permite garantizar el orden de los mensajes por una misma clave.
\n

Productor con serialización de objetos complejos

A menudo es necesario enviar no solo cadenas, sino JSON o datos binarios. Use un serializador personalizado o la biblioteca 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 = "Iván" };\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}

Configuración del consumidor (Consumer)

El consumidor lee mensajes de uno o varios tópicos. Ejemplo de suscripción y procesamiento:

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

Parámetros de configuración importantes:

\n
• GroupId — identificador del grupo de consumidores. Kafka distribuye las particiones entre