Travailler avec Apache Kafka en C++ : bibliothèques et exemples

Travailler avec Apache Kafka en C++ : bibliothèques, configuration et exemples

Apache Kafka est une plateforme distribuée de traitement de flux de données, largement utilisée dans les architectures de microservices, les systèmes de journalisation et d'analyse. Pour l'intégration avec C++, il existe plusieurs bibliothèques matures, dont les plus populaires sont librdkafka (bas niveau) et cppkafka (surcouche C++). Dans cet article, nous allons détailler ces deux bibliothèques, montrer des exemples de code de producteur et de consommateur, et donner des recommandations pour le choix.

1. librdkafka — la base pour travailler avec Kafka en C

librdkafka est une bibliothèque C haute performance de Confluent, implémentant le protocole Apache Kafka. Elle prend en charge toutes les fonctionnalités clés : envoi asynchrone de messages, groupes de consommateurs, sémantique exactly-once (via les transactions) et SSL/SASL. La bibliothèque est écrite en C, mais peut être utilisée depuis C++ directement ou via des surcouches C++.

Installation de librdkafka

Sur Ubuntu/Debian :

sudo apt-get install librdkafka-dev

Sur macOS via Homebrew :

brew install librdkafka

Compilation à partir des sources :

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

Exemple de producteur en C avec librdkafka

Un producteur simple envoyant un message dans un topic :

#include <librdkafka/rdkafka.h>\n#include <iostream>\n#include <string>\n#include <cstring>\n\nint main() {\n    rd_kafka_t *rk;          // gestionnaire du producteur\n    rd_kafka_conf_t *conf;   // configuration\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 << "Échec de la création du producteur : " << errstr << std::endl;\n        return 1;\n    }\n\n    std::string topic_name = "test-topic";\n    std::string message = "Bonjour depuis C++ et 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 << "Échec de l'envoi du message : " << rd_kafka_err2str(err) << std::endl;\n    } else {\n        std::cout << "Message envoyé avec succès !" << std::endl;\n    }\n\n    // Attente de la livraison et nettoyage\n    rd_kafka_flush(rk, 10000);\n    rd_kafka_destroy(rk);\n    return 0;\n}

Exemple de consommateur en C avec librdkafka

Un consommateur simple lisant les messages d'un topic :

#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 << "Échec de la création du consommateur : " << 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 << "En attente de messages..." <<