Trabalhando com Apache Kafka em C# (.NET): biblioteca Confluent.Kafka

Biblioteca Confluent.Kafka para trabalhar com Apache Kafka em C# (.NET)

Apache Kafka é uma plataforma distribuída para streaming de dados, que se tornou padrão na construção de arquiteturas orientadas a eventos e microsserviços. Para desenvolvedores em C# (.NET), a biblioteca mais popular e suportada é a Confluent.Kafka — o cliente oficial da empresa Confluent, construído sobre o cliente C de alto desempenho librdkafka. Neste artigo, abordaremos em detalhes como instalar a biblioteca, criar um produtor e um consumidor, além de explorar as melhores práticas de trabalho.

Instalação do Confluent.Kafka

Para começar, adicione o pacote ao projeto via NuGet:

dotnet add package Confluent.Kafka

Ou através do console do gerenciador de pacotes:

Install-Package Confluent.Kafka

Configuração básica do produtor (Producer)

O produtor envia mensagens para tópicos do Kafka. Veja um exemplo simples:

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

Pontos-chave:

\n
• BootstrapServers — endereço de um ou mais brokers do Kafka.
\n
• ProduceAsync — envio assíncrono com confirmação.
\n
• A Key permite garantir a ordenação das mensagens por uma mesma chave.
\n

Produtor com serialização de objetos complexos

Frequentemente é necessário enviar não apenas strings, mas JSON ou dados binários. Use um serializador personalizado ou a 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 = "João" };\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}

Configuração do consumidor (Consumer)

O consumidor lê mensagens de um ou mais tópicos. Exemplo de inscrição e processamento:

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

Parâmetros de configuração importantes:

\n
• GroupId — identificador do grupo de consumidores. O Kafka distribui as partições entre