العمل مع Apache Kafka في C++: مكتبة librdkafka وأمثلة عملية

مكتبة librdkafka: أداة قوية للعمل مع Kafka في C++

Apache Kafka هي منصة موزعة لتدفق الأحداث أصبحت المعيار الفعلي لبناء أنظمة تبادل الرسائل عالية التحمّل. لدمج Kafka مع C++، المكتبة الأكثر شيوعًا ونضجًا هي librdkafka — مكتبة C عالية الأداء توفر مجموعة كاملة من الإمكانيات للعمل مع منتجي ومستهلكي Kafka. وهي تنفذ بروتوكول Apache Kafka من الإصدار 0.8 فما فوق، بما في ذلك أحدث الإصدارات المستقرة.

الإمكانيات الرئيسية لـ librdkafka

• دعم جميع العمليات الأساسية: إرسال الرسائل (producer) وقراءتها (consumer) مع إدارة الإزاحات (offset).
• العمل مع المواضيع (topics) والأقسام (partitions) ومجموعات المستهلكين.
• دعم أصلي للضغط (gzip، snappy، lz4، zstd).
• تشفير SSL/TLS والمصادقة عبر SASL (PLAIN، SCRAM، GSSAPI، OAUTHBEARER).
• منتج غير مكرر (Idempotent producer) والمعاملات (منذ Kafka 0.11).
• معالجة غير متزامنة مع استدعاءات رجعية (callbacks) وزمن استجابة منخفض.
• ضبط دقيق عبر خصائص التكوين (مشابهة لعميل Java).

التثبيت والربط

يمكن تثبيت librdkafka عبر مدير الحزم أو بناؤها من المصدر. لنظام Linux (Ubuntu/Debian) الأمر:

sudo apt-get install librdkafka-dev

لنظام macOS عبر Homebrew:

brew install librdkafka

بعد التثبيت، قم بتضمين ملف الرأس والمكتبة عند الترجمة:

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

مثال: إنشاء منتج (Producer)

يرسل المنتج الرسائل إلى موضوع (topic) محدد. لنلقِ نظرة على مثال عملي بسيط:

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

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

    // إعداد الوسيط (broker)    conf->set("bootstrap.servers", "localhost:9092", errstr);

    // إنشاء المنتج    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);    if (!producer) {        std::cerr << "Failed to create producer: " << errstr << std::endl;        return 1;    }    delete conf;

    // إرسال رسالة    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;    }

    // انتظار التسليم    producer->flush(10000);    delete producer;    return 0;}

في هذا المثال، نقوم بإنشاء تكوين، وتحديد عنوان الوسيط، وإنشاء مثيل Producer، وإرسال رسالة نصية إلى الموضوع test-topic. تضمن طريقة flush إرسال جميع الرسائل قبل إنهاء البرنامج.

مثال: إنشاء مستهلك (Consumer)

يقرأ المستهلك الرسائل من موضوع ضمن مجموعة مستهلكين. مثال مع إدارة يدوية للإزاحات:

#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