Arbeiten mit Apache Kafka in C++: Die Bibliothek librdkafka und praktische Beispiele

Die Bibliothek librdkafka: Ein leistungsstarkes Werkzeug für die Arbeit mit Kafka in C++

Apache Kafka ist eine verteilte Plattform für Event-Streaming, die sich zum De-facto-Standard für den Aufbau hochlastiger Nachrichtenaustauschsysteme entwickelt hat. Für die Integration von Kafka mit C++ ist die beliebteste und ausgereifteste Bibliothek librdkafka – eine leistungsstarke C-Bibliothek, die einen vollständigen Satz an Funktionen für die Arbeit mit Kafka-Producern und -Consumern bietet. Sie implementiert das Apache-Kafka-Protokoll ab Version 0.8, einschließlich der neuesten stabilen Versionen.

Hauptfunktionen von librdkafka

• Unterstützung aller wichtigen Operationen: Senden von Nachrichten (Producer) und Lesen (Consumer) mit Offset-Verwaltung.
• Arbeit mit Topics, Partitionen und Verbrauchergruppen.
• Native Unterstützung für Komprimierung (gzip, snappy, lz4, zstd).
• SSL/TLS-Verschlüsselung und SASL-Authentifizierung (PLAIN, SCRAM, GSSAPI, OAUTHBEARER).
• Idempotenter Producer und Transaktionen (ab Kafka 0.11).
• Asynchrone Verarbeitung mit Callbacks und geringer Latenz.
• Feinabstimmung über Konfigurationseigenschaften (analog zum Java-Client).

Installation und Einbindung

Sie können librdkafka über den Paketmanager installieren oder aus den Quellen kompilieren. Für Linux (Ubuntu/Debian) lautet der Befehl:

sudo apt-get install librdkafka-dev

Für macOS über Homebrew:

brew install librdkafka

Nach der Installation binden Sie die Header-Datei und die Bibliothek bei der Kompilierung ein:

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

Beispiel: Erstellen eines Producers

Der Producer sendet Nachrichten an ein bestimmtes Topic. Betrachten wir ein minimales funktionierendes Beispiel:

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

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

    // Broker-Konfiguration    conf->set("bootstrap.servers", "localhost:9092", errstr);

    // Producer erstellen    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);    if (!producer) {        std::cerr << "Fehler beim Erstellen des Producers: " << errstr << std::endl;        return 1;    }    delete conf;

    // Nachricht senden    std::string topic_name = "test-topic";    std::string payload = "Hallo von 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 << "Senden fehlgeschlagen: " << RdKafka::err2str(resp) << std::endl;    } else {        std::cout << "Nachricht erfolgreich gesendet" << std::endl;    }

    // Auf Zustellung warten    producer->flush(10000);    delete producer;    return 0;}

In diesem Beispiel erstellen wir eine Konfiguration, geben die Broker-Adresse an, erstellen eine Producer-Instanz und senden eine Zeichenfolgennachricht an das Topic test-topic. Die Methode flush stellt sicher, dass alle Nachrichten vor dem Beenden des Programms gesendet werden.

Beispiel: Erstellen eines Consumers

Der Consumer liest Nachrichten aus einem Topic innerhalb einer Verbrauchergruppe. Ein Beispiel mit manueller Offset-Verwaltung:

#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 << "Fehler: " << msg.errs