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

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

Apache Kafka é uma plataforma distribuída para streaming de dados, amplamente utilizada para construir sistemas de troca de mensagens de alta carga, processamento de eventos e integração de microsserviços. Para trabalhar com Kafka no ecossistema .NET, a principal e mais suportada biblioteca é a Confluent.Kafka — o cliente oficial da Confluent, construído sobre a biblioteca C de alto desempenho librdkafka.

Neste artigo, abordaremos os principais recursos do Confluent.Kafka, instalação, configuração do produtor e consumidor, tratamento de erros e exemplos práticos de código em C#.

Instalação e configuração

A biblioteca é distribuída via NuGet. Para instalar, execute o comando:

dotnet add package Confluent.Kafka

Após a instalação, adicione as diretivas using ao seu código:

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

Conceitos básicos

• Producer (Produtor) — envia mensagens para tópicos do Kafka.
• Consumer (Consumidor) — lê mensagens dos tópicos.
• Topic (Tópico) — canal nomeado onde as mensagens são publicadas.
• Broker (Broker) — servidor Kafka que armazena os dados.
• Partition (Partição) — unidade de paralelismo dentro de um tópico.

Criando um produtor (Producer)

O produtor envia mensagens para o Kafka. Veja um exemplo 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 = "Olá Kafka do .NET!" }        );

        Console.WriteLine($"Enviado para partition [{deliveryResult.Partition}] no offset [{deliveryResult.Offset}]");    }    catch (ProduceException<string, string> e)    {        Console.WriteLine($"Erro ao enviar: {e.Error.Reason}");    }}

Neste exemplo:

• BootstrapServers — endereço do cluster Kafka.
• ProduceAsync — envio assíncrono da mensagem.
• Key — chave opcional para determinar a partição.
• Value — corpo da mensagem (pode ser JSON, string, dados binários).

Criando um consumidor (Consumer)

O consumidor se inscreve em tópicos e processa as mensagens. Um parâmetro importante é o GroupId, que define o 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($"Mensagem recebida: Key = {consumeResult.Message.Key}, Value = {consumeResult.Message.Value}");

            // Confirmação manual (commit)            consumer.Commit(consumeResult);        }    }    catch (OperationCanceledException)    {        Console.WriteLine("Consumo interrompido.");    }    finally    {        consumer.Close();    }}

Pontos-chave:

• AutoOff