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

مكتبة KafkaJS: العمل مع Apache Kafka في TypeScript

\n\n

Apache Kafka هي منصة موزعة لمعالجة تدفق البيانات، تُستخدم على نطاق واسع لبناء أنظمة تبادل الرسائل عالية التحميل. في نظام TypeScript (و Node.js) البيئي، أصبحت مكتبة KafkaJS المعيار الفعلي. توفر واجهة برمجة تطبيقات (API) نظيفة وحديثة مع دعم كامل لـ TypeScript، مما يسمح للمطورين بكتابة كود آمن من حيث الأنواع للمنتجين والمستهلكين في Kafka.

\n\n

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

\n\n

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

\n\n

للبدء، قم بتثبيت المكتبة عبر npm أو yarn:

\n\n

npm install kafkajs\n# أو\nyarn add kafkajs

\n\n

KafkaJS مكتوبة بلغة TypeScript، لذا فإن الأنواع مضمنة بالفعل — لا حاجة لتثبيت @types/kafkajs بشكل إضافي.

\n\n

الآن، لننشئ مثيلاً لعميل Kafka. يُوصى بتحديد عدة وسطاء (brokers) لتحمل الأخطاء:

\n\n

import { Kafka, logLevel } from 'kafkajs';\n\nconst kafka = new Kafka({\n  clientId: 'my-app',\n  brokers: ['localhost:9092', 'localhost:9093'],\n  logLevel: logLevel.INFO, // مستوى التسجيل\n  retry: {\n    initialRetryTime: 100,\n    retries: 8\n  }\n});

\n\n

المنتج: إرسال الرسائل إلى Kafka

\n\n

المنتج (producer) مسؤول عن كتابة الرسائل إلى مواضيع (topics) Kafka. توفر KafkaJS واجهة برمجة تطبيقات (API) ملائمة لإرسال الرسائل الفردية والمجمعة (batches).

\n\n

مثال على منتج أساسي:

\n\n

import { Kafka, Producer } from 'kafkajs';\n\nasync function produceMessage(producer: Producer) {\n  await producer.send({\n    topic: 'orders',\n    messages: [\n      { \n        key: 'order-123', \n        value: JSON.stringify({ userId: 42, amount: 1500 }) \n      },\n    ],\n  });\n  console.log('تم إرسال الرسالة');\n}\n\nasync function runProducer() {\n  const producer = kafka.producer();\n  await producer.connect();\n\n  // إرسال عدة رسائل\n  await producer.send({\n    topic: 'orders',\n    messages: [\n      { key: 'order-124', value: '{"item":"laptop"}' },\n      { key: 'order-125', value: '{"item":"mouse"}' },\n    ],\n  });\n\n  await produceMessage(producer);\n  await producer.disconnect();\n}\n\nrunProducer().catch(console.error);

\n\n

هام: استدعِ دائمًا disconnect() بعد الانتهاء من العمل لتحرير الموارد. بالنسبة للتطبيقات طويلة العمر، يمكن إبقاء المنتج متصلاً بشكل دائم.

\n\n

المستهلك: قراءة الرسائل من Kafka

\n\n

المستهلك (consumer) يشترك في موضوع واحد أو أكثر ويعالج الرسائل. تدعم KafkaJS مجموعات المستهلكين للمعالجة المتوازية.

\n\n

مثال على مستهلك مع معالجة الرسائل:

\n\n

import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';\n\nasync function runConsumer() {\n  const consumer: Consumer = kafka.consumer({ \n    groupId: 'order-processor-group' \n  });\n\n  await consumer.connect();\n  await consumer.subscribe({ \n    topic: 'orders', \n    fromBeginning: true // القراءة من أول رسالة\n  });\n\n  await consumer.run({\n    eachMessage: async ({ topic, partition, message }: EachMessagePayload) => {\n      const key = message.key?.toString();\n      const value = message.value?.toString();\n      console.log({\n        partition,\n        offset: message.offset,\n        key,\n        value: value ? JSON.parse(value) : null,\n      });\n    },\n  });\n}\n\nrunConsumer().catch(console.error);

\n\n

لاحظ المعامل fromBeginning: إذا تم تعيينه على true، سيقرأ المستهلك جميع الرسائل المتراكمة سابقًا. لبيئة الإنتاج، يُستخدم عادةً false (الرسائل الجديدة فقط).