Trabalhando com Apache Kafka em C++: bibliotecas e exemplos

Trabalhando com Apache Kafka em C++: bibliotecas, configuração e exemplos

Apache Kafka é uma plataforma distribuída para processamento de fluxo de dados, amplamente utilizada em arquiteturas de microsserviços, sistemas de logging e análise. Para integração com C++, existem várias bibliotecas maduras, das quais as mais populares são librdkafka (de baixo nível) e cppkafka (wrapper em C++). Neste artigo, detalharemos ambas as bibliotecas, mostraremos exemplos de código de produtor e consumidor e daremos recomendações de escolha.

1. librdkafka — base para trabalhar com Kafka em C

librdkafka é uma biblioteca C de alto desempenho da Confluent que implementa o protocolo Apache Kafka. Ela suporta todos os recursos principais: envio assíncrono de mensagens, grupos de consumidores, semântica exactly-once (através de transações) e SSL/SASL. A biblioteca é escrita em C, mas pode ser usada diretamente de C++ ou através de wrappers C++.

Instalação do librdkafka

No Ubuntu/Debian:

sudo apt-get install librdkafka-dev

No macOS via Homebrew:

brew install librdkafka

Compilação a partir do código fonte:

git clone https://github.com/confluentinc/librdkafka.git\ncd librdkafka\n./configure --prefix=/usr/local\nmake\nsudo make install

Exemplo de produtor em C usando librdkafka

Produtor simples, enviando uma mensagem para um tópico:

#include <librdkafka/rdkafka.h>\n#include <iostream>\n#include <string>\n#include <cstring>\n\nint main() {\n    rd_kafka_t *rk;          // handler do produtor\n    rd_kafka_conf_t *conf;   // configuração\n    char errstr[512];\n\n    conf = rd_kafka_conf_new();\n    rd_kafka_conf_set(conf, "bootstrap.servers", "localhost:9092", errstr, sizeof(errstr));\n    rd_kafka_conf_set(conf, "acks", "all", errstr, sizeof(errstr));\n\n    rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));\n    if (!rk) {\n        std::cerr << "Falha ao criar produtor: " << errstr << std::endl;\n        return 1;\n    }\n\n    std::string topic_name = "test-topic";\n    std::string message = "Olá do C++ e librdkafka!";\n\n    rd_kafka_resp_err_t err = rd_kafka_producev(\n        rk,\n        RD_KAFKA_V_TOPIC(topic_name.c_str()),\n        RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),\n        RD_KAFKA_V_VALUE((void*)message.c_str(), message.size()),\n        RD_KAFKA_V_END\n    );\n\n    if (err) {\n        std::cerr << "Falha ao produzir mensagem: " << rd_kafka_err2str(err) << std::endl;\n    } else {\n        std::cout << "Mensagem enviada com sucesso!" << std::endl;\n    }\n\n    // Aguardar entrega e limpeza\n    rd_kafka_flush(rk, 10000);\n    rd_kafka_destroy(rk);\n    return 0;\n}

Exemplo de consumidor em C com librdkafka

Consumidor simples, lendo mensagens de um tópico:

#include <librdkafka/rdkafka.h>\n#include <iostream>\n#include <string>\n\nint main() {\n    rd_kafka_t *rk;\n    rd_kafka_conf_t *conf;\n    rd_kafka_topic_partition_list_t *topics;\n    char errstr[512];\n\n    conf = rd_kafka_conf_new();\n    rd_kafka_conf_set(conf, "bootstrap.servers", "localhost:9092", errstr, sizeof(errstr));\n    rd_kafka_conf_set(conf, "group.id", "my-group", errstr, sizeof(errstr));\n    rd_kafka_conf_set(conf, "auto.offset.reset", "earliest", errstr, sizeof(errstr));\n\n    rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));\n    if (!rk) {\n        std::cerr << "Falha ao criar consumidor: " << errstr << std::endl;\n        return 1;\n    }\n\n    rd_kafka_poll_set_consumer(rk);\n    topics = rd_kafka_topic_partition_list_new(1);\n    rd_kafka_topic_partition_list_add(topics, "test-topic", RD_KAFKA_PARTITION_UA);\n    rd_kafka_consumer_subscribe(rk, topics);\n\n    std::cout << "Aguardando mensagens..." <<