العمل مع Apache Kafka في Node.js: مكتبة KafkaJS

العمل مع Apache Kafka في Node.js باستخدام مكتبة KafkaJS

\n\n

Apache Kafka هي منصة موزعة لتدفق الأحداث، أصبحت المعيار الفعلي لبناء أنظمة تبادل الرسائل عالية التحميل. في نظام Node.js، واحدة من أشهر وأكثر المكتبات استقرارًا للتفاعل مع Kafka هي KafkaJS. وهي عميل حديث وغير متزامن بالكامل، مكتوب بلغة JavaScript النقية، ويدعم جميع إمكانيات Kafka الأساسية: المنتجين، المستهلكين، مجموعات المستهلكين، إدارة الإزاحات (offsets)، والمعاملات.

\n\n

في هذه المقالة، سنشرح بالتفصيل كيفية تثبيت KafkaJS، وإعداد الاتصال بمجموعة Kafka، وإرسال وقراءة الرسائل، بالإضافة إلى أفضل الممارسات للعمل مع هذه المكتبة في بيئة الإنتاج.

\n\n

تثبيت وإعداد KafkaJS

\n\n

يتم تثبيت المكتبة باستخدام مدير الحزم القياسي npm:

\n\n

npm install kafkajs

\n\n

بعد التثبيت، يجب إنشاء عميل مع تحديد عناوين الوسطاء (brokers) ومعرف العميل. مثال على الحد الأدنى من التكوين:

\n\n

const { Kafka } = require('kafkajs');\n\nconst kafka = new Kafka({\n  clientId: 'my-app',\n  brokers: ['localhost:9092']\n});

\n\n

في بيئة الإنتاج، يُوصى بتحديد عدة وسطاء مفصولين بفواصل لضمان تحمل الأخطاء، بالإضافة إلى إعداد SSL ومصادقة SASL:

\n\n

const kafka = new Kafka({\n  clientId: 'my-production-app',\n  brokers: ['broker1:9092', 'broker2:9092', 'broker3:9092'],\n  ssl: true,\n  sasl: {\n    mechanism: 'scram-sha-256',\n    username: 'myuser',\n    password: 'mypassword'\n  }\n});

\n\n

المنتج: إرسال الرسائل إلى الموضوع (topic)

\n\n

المنتج مسؤول عن نشر الرسائل في موضوع Kafka المحدد. توفر KafkaJS وضعين لتشغيل المنتج: بسيط (send) ومعاملاتي (للتوصيل الدقيق). دعنا نلقي نظرة على مثال أساسي لإرسال رسالة:

\n\n

const producer = kafka.producer();\n\nasync function sendMessage() {\n  await producer.connect();\n\n  await producer.send({\n    topic: 'test-topic',\n    messages: [\n      { key: 'user1', value: 'Hello Kafka from Node.js!' },\n      { key: 'user2', value: JSON.stringify({ orderId: 123, amount: 2500 }) }\n    ]\n  });\n\n  await producer.disconnect();\n}\n\nsendMessage().catch(console.error);

\n\n

النقاط الرئيسية:

\n

\n
• topic — اسم الموضوع الذي سيتم إرسال الرسالة إليه.
\n
• messages — مصفوفة من الكائنات، يمكن أن يحتوي كل منها على key (اختياري) و value (سلسلة نصية أو Buffer).
\n
• يُوصى دائمًا باستدعاء producer.connect() مرة واحدة عند بدء التطبيق وإعادة استخدام الاتصال.
\n

\n\n

لتحسين الأداء في الإنتاج، استخدم تجمع الاتصالات وإعداد acks:

\n\n

const producer = kafka.producer({\n  allowAutoTopicCreation: false,\n  transactionTimeout: 30000\n});\n\nawait producer.send({\n  topic: 'orders',\n  messages: [{ value: 'New order' }],\n  acks: -1  // انتظار التأكيد من جميع النسخ المتماثلة\n});

\n\n

المستهلك: قراءة الرسائل من الموضوع

\n\n

يسمح المستهلك بقراءة الرسائل من موضوع واحد أو أكثر. المفهوم الأكثر أهمية هو مجموعات المستهلكين (consumer groups)، التي تضمن موازنة الحمل وتضمن معالجة كل رسالة بواسطة مستهلك واحد فقط من المجموعة.

\n\n

const consumer = kafka.consumer({ groupId: 'my-group' });\n\nasync function consumeMessages() {\n  await consumer.connect();\n  await consumer.subscribe({ topic: 'test-topic', fromBeginning: true });\n\n  await consumer.run({\n    eachMessage: async ({ topic, partition, message }) => {\n      console.log({\n        partition,\n        offset: message.offset,\n        key: message.key.toString(),\n        value: message.valu