Trabalhando com Apache Kafka em C++: biblioteca librdkafka e exemplos práticos

Biblioteca librdkafka: ferramenta poderosa para trabalhar com Kafka em C++

Apache Kafka é uma plataforma distribuída para streaming de eventos que se tornou o padrão de facto para a construção de sistemas de troca de mensagens de alta carga. Para integrar Kafka com C++, a biblioteca mais popular e madura é a librdkafka — uma biblioteca C de alto desempenho que oferece um conjunto completo de recursos para trabalhar com produtores e consumidores Kafka. Ela implementa o protocolo Apache Kafka versões 0.8 e superiores, incluindo as versões estáveis mais recentes.

Principais recursos da librdkafka

• Suporte a todas as operações principais: envio de mensagens (producer) e leitura (consumer) com gerenciamento de offsets.
• Trabalho com tópicos, partições e grupos de consumidores.
• Suporte nativo a compressão (gzip, snappy, lz4, zstd).
• Criptografia SSL/TLS e autenticação SASL (PLAIN, SCRAM, GSSAPI, OAUTHBEARER).
• Produtor idempotente e transações (a partir do Kafka 0.11).
• Processamento assíncrono com callbacks e baixa latência.
• Configuração refinada através de propriedades de configuração (semelhante ao cliente Java).

Instalação e configuração

Você pode instalar a librdkafka através do gerenciador de pacotes ou compilar a partir do código-fonte. Para Linux (Ubuntu/Debian), o comando é:

sudo apt-get install librdkafka-dev

Para macOS via Homebrew:

brew install librdkafka

Após a instalação, inclua o arquivo de cabeçalho e a biblioteca durante a compilação:

g++ -o my_app my_app.cpp -lrdkafka -lpthread -lz

Exemplo: criando um produtor (Producer)

O produtor envia mensagens para um tópico especificado. Vejamos um exemplo funcional mínimo:

#include <iostream>#include <string>#include <librdkafka/rdkafkacpp.h>

int main() {    std::string errstr;    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

    // Configuração do broker    conf->set("bootstrap.servers", "localhost:9092", errstr);

    // Criação do produtor    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);    if (!producer) {        std::cerr << "Falha ao criar produtor: " << errstr << std::endl;        return 1;    }    delete conf;

    // Envio de mensagem    std::string topic_name = "test-topic";    std::string payload = "Olá do C++ Kafka!";    RdKafka::ErrorCode resp = producer->produce(        topic_name,        RdKafka::Topic::PARTITION_UA,        RdKafka::Producer::RK_MSG_COPY,        const_cast<char*>(payload.c_str()),        payload.size(),        nullptr, 0, 0, nullptr    );

    if (resp != RdKafka::ERR_NO_ERROR) {        std::cerr << "Falha ao produzir: " << RdKafka::err2str(resp) << std::endl;    } else {        std::cout << "Mensagem enviada com sucesso" << std::endl;    }

    // Aguardando entrega    producer->flush(10000);    delete producer;    return 0;}

Neste exemplo, criamos uma configuração, especificamos o endereço do broker, criamos uma instância de Producer e enviamos uma mensagem de string para o tópico test-topic. O método flush garante que todas as mensagens sejam enviadas antes do término do programa.

Exemplo: criando um consumidor (Consumer)

O consumidor lê mensagens de um tópico como parte de um grupo de consumidores. Exemplo com gerenciamento manual de offsets:

#include <iostream>#include <string>#include <librdkafka/rdkafkacpp.h>

class MyConsumerCb : public RdKafka::ConsumeCb {public:    void consume_cb(RdKafka::Message &msg, void *opaque) override {        if (msg.err()) {            if (msg.err() != RdKafka::ERR__TIMED_OUT) {                std::cerr << "Erro: " << msg.errs