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

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

Apache Kafka هي منصة موزعة لمعالجة البيانات المتدفقة، تُستخدم على نطاق واسع في بنية الخدمات المصغرة وأنظمة التسجيل والتحليلات. للتكامل مع C++، توجد عدة مكتبات ناضجة، أشهرها librdkafka (مستوى منخفض) و cppkafka (غلاف بلغة C++). في هذه المقالة، سنشرح بالتفصيل كلتا المكتبتين، ونعرض أمثلة على كود المنتج والمستهلك، ونقدم توصيات للاختيار.

1. librdkafka - أساس العمل مع Kafka بلغة C

librdkafka هي مكتبة C عالية الأداء من Confluent، تنفذ بروتوكول Apache Kafka. تدعم جميع الوظائف الرئيسية: الإرسال غير المتزامن للرسائل، مجموعات المستهلكين، دلالة مرة واحدة بالضبط (عبر المعاملات) و SSL/SASL. المكتبة مكتوبة بلغة C، ولكن يمكن استخدامها من C++ مباشرة أو عبر أغلفة C++.

تثبيت librdkafka

على Ubuntu/Debian:

sudo apt-get install librdkafka-dev

على macOS عبر Homebrew:

brew install librdkafka

البناء من المصدر:

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

مثال على منتج بلغة C باستخدام librdkafka

منتج بسيط يرسل رسالة إلى موضوع:

#include <librdkafka/rdkafka.h>\n#include <iostream>\n#include <string>\n#include <cstring>\n\nint main() {\n    rd_kafka_t *rk;          // معالج المنتج\n    rd_kafka_conf_t *conf;   // الإعدادات\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 << "فشل إنشاء المنتج: " << errstr << std::endl;\n        return 1;\n    }\n\n    std::string topic_name = "test-topic";\n    std::string message = "مرحباً من C++ و 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 << "فشل إرسال الرسالة: " << rd_kafka_err2str(err) << std::endl;\n    } else {\n        std::cout << "تم إرسال الرسالة بنجاح!" << std::endl;\n    }\n\n    // انتظار التسليم والتنظيف\n    rd_kafka_flush(rk, 10000);\n    rd_kafka_destroy(rk);\n    return 0;\n}

مثال على مستهلك بلغة C باستخدام librdkafka

مستهلك بسيط يقرأ الرسائل من موضوع:

#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 << "فشل إنشاء المستهلك: " << 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 << "في انتظار الرسائل..." <<