العمل مع Kafka في PHP: المكتبات، الاتصال وأمثلة الكود

العمل مع Kafka في PHP: المكتبات، الاتصال وأمثلة الكود

\n\n

Apache Kafka هي منصة موزعة لتدفق البيانات تتيح نشر وتخزين ومعالجة تدفقات الرسائل في الوقت الفعلي. في نظام PHP البيئي، يُستخدم امتداد php-rdkafka للعمل مع Kafka - وهو غلاف لمكتبة librdkafka المكتوبة بلغة C. في هذه المقالة، سنشرح كيفية تثبيت الأداة، كتابة منتج ومستهلك، بالإضافة إلى استعراض السيناريوهات النموذجية للاستخدام.

\n\n

ما هو php-rdkafka؟

\n\n

php-rdkafka هو امتداد PHP رسمي للعمل مع Apache Kafka. يوفر واجهة عالية الأداء لإرسال واستقبال الرسائل. يدعم الامتداد جميع الإمكانيات الرئيسية لـ Kafka: المواضيع، الأقسام، مجموعات المستهلكين، الالتزامات اليدوية، الضغط وغيرها الكثير.

\n\n

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

\n

\n
• إنشاء منتجين (producers) لإرسال الرسائل إلى المواضيع.
\n
• إنشاء مستهلكين (consumers) لقراءة الرسائل من المواضيع.
\n
• دعم مجموعات المستهلكين وإعادة التوازن التلقائي.
\n
• العمل مع الأقسام والتحكم اليدوي في الإزاحات (offsets).
\n
• ضغط الرسائل (gzip, snappy, lz4, zstd).
\n
• الإرسال غير المتزامن والمعالجة المجمعة.
\n

\n\n

تثبيت php-rdkafka

\n\n

يمكن تثبيت الامتداد عبر PECL أو البناء من المصدر. الطريقة الموصى بها هي عبر PECL:

\n\n

# التثبيت عبر PECL\npecl install rdkafka\n\n# لنظام Ubuntu/Debian قم بتثبيت librdkafka-dev أولاً\nsudo apt-get install librdkafka-dev\npecl install rdkafka\n\n# أضف الامتداد إلى ملف php.ini\nextension=rdkafka.so

\n\n

بعد التثبيت، تحقق من تحميل الامتداد:

\n\n

php -m | grep rdkafka

\n\n

إذا كنت تستخدم Composer، يمكنك أيضًا تثبيت الحزمة arnaud-lb/php-rdkafka للتحميل التلقائي:

\n\n

composer require arnaud-lb/php-rdkafka

\n\n

مثال على المنتج (Producer)

\n\n

يرسل المنتج الرسائل إلى موضوع Kafka محدد. لنستعرض مثالًا أساسيًا مع الإرسال المتزامن:

\n\n

<?php\nrequire_once 'vendor/autoload.php';\n\n$conf = new RdKafka\\Conf();\n$conf->set('metadata.broker.list', 'localhost:9092');\n\n$producer = new RdKafka\\Producer($conf);\n$topic = $producer->newTopic('test_topic');\n\n// إرسال رسالة\n$topic->produce(RD_KAFKA_PARTITION_UA, 0, 'Hello, Kafka from PHP!');\n\n// انتظار إرسال جميع الرسائل\n$producer->flush(1000);\necho "تم إرسال الرسالة\\n";

\n\n

شرح الكود:

\n

\n
• metadata.broker.list — عنوان وسيط Kafka (افتراضيًا localhost:9092).
\n
• RD_KAFKA_PARTITION_UA — اختيار تلقائي للقسم.
\n
• flush(1000) — يمنع التنفيذ حتى إرسال جميع الرسائل (مهلة ثانية واحدة).
\n

\n\n

مثال على المستهلك (Consumer)

\n\n

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

\n\n

<?php\nrequire_once 'vendor/autoload.php';\n\n$conf = new RdKafka\\Conf();\n$conf->set('metadata.broker.list', 'localhost:9092');\n$conf->set('group.id', 'my_consumer_group');\n$conf->set('auto.offset.reset', 'earliest');\n\n$consumer = new RdKafka\\KafkaConsumer($conf);\n$consumer->subscribe(['test_topic']);\n\necho "انتظار الرسائل...\\n";\n\nwhile (true) {\n    $message = $consumer->consume(120 * 1000); // مهلة دقيقتين\n    switch ($message->err) {\n        case RD_KAFKA_RESP_ERR_NO_ERROR:\n            echo "تم استلام الرسالة: " . $message->payload . "\\n";\n            break;\n        case RD_KAFKA_RESP_ERR__PARTITION_EOF:\n            echo "تم الوصول إلى نهاية القسم\\n";\n            break;\n        case RD_KAFKA_RESP_ERR__TIMED_OUT:\n            echo "انتهت المهلة\\n";\n            break;\n        defau