Trabajo con Apache Kafka en C++: bibliotecas y ejemplos

Trabajo con Apache Kafka en C++: bibliotecas, configuración y ejemplos

Apache Kafka es una plataforma distribuida para el procesamiento de flujos de datos, ampliamente utilizada en arquitecturas de microservicios, sistemas de registro y análisis. Para la integración con C++, existen varias bibliotecas maduras, de las cuales las más populares son librdkafka (de bajo nivel) y cppkafka (envoltura en C++). En este artículo, analizaremos ambas bibliotecas en detalle, mostraremos ejemplos de código de productor y consumidor, y daremos recomendaciones para su elección.

1. librdkafka — base para trabajar con Kafka en C

librdkafka es una biblioteca C de alto rendimiento de Confluent que implementa el protocolo Apache Kafka. Soporta todas las funciones clave: envío asíncrono de mensajes, grupos de consumidores, semántica exactly-once (a través de transacciones) y SSL/SASL. La biblioteca está escrita en C, pero puede usarse desde C++ directamente o a través de envolturas C++.

Instalación de librdkafka

En Ubuntu/Debian:

sudo apt-get install librdkafka-dev

En macOS con Homebrew:

brew install librdkafka

Compilación desde el código fuente:

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

Ejemplo de productor en C con librdkafka

Productor simple que envía un mensaje a un tópico:

#include <librdkafka/rdkafka.h>\n#include <iostream>\n#include <string>\n#include <cstring>\n\nint main() {\n    rd_kafka_t *rk;          // manejador del productor\n    rd_kafka_conf_t *conf;   // configuración\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 << "Failed to create producer: " << errstr << std::endl;\n        return 1;\n    }\n\n    std::string topic_name = "test-topic";\n    std::string message = "Hello from C++ and 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 << "Failed to produce message: " << rd_kafka_err2str(err) << std::endl;\n    } else {\n        std::cout << "Message sent successfully!" << std::endl;\n    }\n\n    // Esperar entrega y limpiar\n    rd_kafka_flush(rk, 10000);\n    rd_kafka_destroy(rk);\n    return 0;\n}

Ejemplo de consumidor en C con librdkafka

Consumidor simple que lee mensajes de un 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 << "Failed to create consumer: " << 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 << "Waiting for messages..." <<