Travailler avec Apache Kafka en C++ : la bibliothèque librdkafka et exemples pratiques

Bibliothèque librdkafka : un outil puissant pour travailler avec Kafka en C++

Apache Kafka est une plateforme distribuée de streaming d'événements qui est devenue le standard de facto pour la construction de systèmes d'échange de messages à forte charge. Pour intégrer Kafka avec C++, la bibliothèque la plus populaire et la plus mature est librdkafka — une bibliothèque C haute performance offrant un ensemble complet de fonctionnalités pour travailler avec les producteurs et consommateurs Kafka. Elle implémente le protocole Apache Kafka versions 0.8 et supérieures, y compris les dernières versions stables.

Principales fonctionnalités de librdkafka

• Prise en charge de toutes les opérations clés : envoi de messages (producer) et lecture (consumer) avec gestion des offsets.
• Travail avec les topics, partitions et groupes de consommateurs.
• Prise en charge native de la compression (gzip, snappy, lz4, zstd).
• Chiffrement SSL/TLS et authentification SASL (PLAIN, SCRAM, GSSAPI, OAUTHBEARER).
• Producteur idempotent et transactions (à partir de Kafka 0.11).
• Traitement asynchrone avec callbacks et faible latence.
• Configuration fine via des propriétés de configuration (similaire au client Java).

Installation et connexion

Vous pouvez installer librdkafka via un gestionnaire de paquets ou le compiler à partir des sources. Pour Linux (Ubuntu/Debian), la commande :

sudo apt-get install librdkafka-dev

Pour macOS via Homebrew :

brew install librdkafka

Après l'installation, incluez le fichier d'en-tête et la bibliothèque lors de la compilation :

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

Exemple : création d'un producteur (Producer)

Le producteur envoie des messages dans un topic spécifié. Voici un exemple fonctionnel minimal :

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

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

    // Configuration du broker    conf->set("bootstrap.servers", "localhost:9092", errstr);

    // Création du producteur    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);    if (!producer) {        std::cerr << "Failed to create producer: " << errstr << std::endl;        return 1;    }    delete conf;

    // Envoi du message    std::string topic_name = "test-topic";    std::string payload = "Hello from 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 << "Produce failed: " << RdKafka::err2str(resp) << std::endl;    } else {        std::cout << "Message sent successfully" << std::endl;    }

    // Attente de la livraison    producer->flush(10000);    delete producer;    return 0;}

Dans cet exemple, nous créons une configuration, spécifions l'adresse du broker, créons une instance Producer et envoyons un message textuel dans le topic test-topic. La méthode flush garantit que tous les messages sont envoyés avant la fin du programme.

Exemple : création d'un consommateur (Consumer)

Le consommateur lit les messages d'un topic au sein d'un groupe de consommateurs. Exemple avec gestion manuelle des 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 << "Error: " << msg.errs