Arbeit mit Apache Kafka in C++: Bibliotheken und Beispiele

Arbeit mit Apache Kafka in C++: Bibliotheken, Konfiguration und Beispiele

Apache Kafka ist eine verteilte Plattform für die Stream-Verarbeitung von Daten, die häufig in Microservice-Architekturen, Logging-Systemen und Analysen eingesetzt wird. Für die Integration mit C++ gibt es mehrere ausgereifte Bibliotheken, von denen librdkafka (niedrige Ebene) und cppkafka (C++-Wrapper) am beliebtesten sind. In diesem Artikel werden wir beide Bibliotheken im Detail besprechen, Codebeispiele für Producer und Consumer zeigen und Empfehlungen zur Auswahl geben.

1. librdkafka – Grundlage für die Arbeit mit Kafka in C

librdkafka ist eine leistungsstarke C-Bibliothek von Confluent, die das Apache Kafka-Protokoll implementiert. Sie unterstützt alle wichtigen Funktionen: asynchrones Senden von Nachrichten, Verbrauchergruppen, Exactly-Once-Semantik (über Transaktionen) und SSL/SASL. Die Bibliothek ist in C geschrieben, kann aber direkt von C++ oder über C++-Wrapper verwendet werden.

Installation von librdkafka

Auf Ubuntu/Debian:

sudo apt-get install librdkafka-dev

Auf macOS über Homebrew:

brew install librdkafka

Build aus dem Quellcode:

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

Beispiel für einen Producer in C mit librdkafka

Ein einfacher Producer, der eine Nachricht an ein Topic sendet:

#include <librdkafka/rdkafka.h>\n#include <iostream>\n#include <string>\n#include <cstring>\n\nint main() {\n    rd_kafka_t *rk;          // Producer-Handle\n    rd_kafka_conf_t *conf;   // Konfiguration\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 << "Fehler beim Erstellen des Producers: " << errstr << std::endl;\n        return 1;\n    }\n\n    std::string topic_name = "test-topic";\n    std::string message = "Hallo von C++ und 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 << "Fehler beim Senden der Nachricht: " << rd_kafka_err2str(err) << std::endl;\n    } else {\n        std::cout << "Nachricht erfolgreich gesendet!" << std::endl;\n    }\n\n    // Warten auf Zustellung und Bereinigung\n    rd_kafka_flush(rk, 10000);\n    rd_kafka_destroy(rk);\n    return 0;\n}

Beispiel für einen Consumer in C mit librdkafka

Ein einfacher Consumer, der Nachrichten aus einem Topic liest:

#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 << "Fehler beim Erstellen des Consumers: " << 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 << "Warte auf Nachrichten..." <<