العمل مع Kafka في TypeScript: أمثلة KafkaJS

العمل مع Kafka في TypeScript باستخدام KafkaJS

Apache Kafka هي منصة موزعة لتدفق البيانات، تُستخدم على نطاق واسع في بنية الخدمات المصغرة، ومعالجة الأحداث، والتحليلات في الوقت الفعلي. عند التطوير باستخدام TypeScript، واحدة من أشهر المكتبات للتفاعل مع Kafka هي KafkaJS. توفر واجهة برمجة تطبيقات حديثة وغير متزامنة مع دعم كامل لـ TypeScript، مما يجعلها الخيار الأمثل لبناء تطبيقات موثوقة وقابلة للتوسع.

ما هي KafkaJS؟

KafkaJS هي مكتبة عميل لـ Apache Kafka، مكتوبة بلغة JavaScript/TypeScript. تدعم جميع العمليات الأساسية: إرسال وقراءة الرسائل، إدارة مجموعات المستهلكين، العمل مع المواضيع والأقسام. بفضل الأنواع المدمجة، توفر KafkaJS كتابة صارمة وإكمال تلقائي في بيئة التطوير المتكاملة (IDE)، مما يبسط التطوير وتصحيح الأخطاء بشكل كبير.

تثبيت KafkaJS

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

npm install kafkajs

إذا كنت تستخدم TypeScript، فلا حاجة لأنواع إضافية - فهي مضمنة بالفعل في الحزمة.

الإعداد الأساسي والاتصال

الخطوة الأولى هي إنشاء مثيل لعميل Kafka. حدد معرف العميل وعناوين الوسطاء (brokers):

import { Kafka } from 'kafkajs';

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

// التحقق من الاتصالconst admin = kafka.admin();await admin.connect();console.log('Connected to Kafka');await admin.disconnect();

إنشاء منتج (Producer)

المنتج يرسل الرسائل إلى المواضيع. تدعم KafkaJS الإنشاء التلقائي للمواضيع عند الإرسال (إذا كان مفعلاً في تكوين الوسيط).

import { Kafka, Partitioners } from 'kafkajs';

const producer = kafka.producer({  createPartitioner: Partitioners.DefaultPartitioner});

await producer.connect();

// إرسال رسالة واحدةawait producer.send({  topic: 'orders',  messages: [    { key: 'order-123', value: JSON.stringify({ userId: 1, amount: 250 }) }  ]});

// إرسال عدة رسائلawait producer.send({  topic: 'orders',  messages: [    { key: 'order-124', value: 'Message 2' },    { key: 'order-125', value: 'Message 3' }  ]});

await producer.disconnect();

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

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

const consumer = kafka.consumer({ groupId: 'order-group' });

await consumer.connect();await consumer.subscribe({ topic: 'orders', fromBeginning: true });

await consumer.run({  eachMessage: async ({ topic, partition, message }) => {    console.log({      partition,      offset: message.offset,      key: message.key?.toString(),      value: message.value?.toString(),    });  },});

العمل مع مجموعات المستهلكين

مجموعات المستهلكين تسمح بتوزيع الحمل بين عدة نسخ من المستهلك. يقوم Kafka تلقائياً بتعيين أقسام الموضوع لمستهلكين مختلفين داخل نفس المجموعة. إذا تعطل أحد المستهلكين، يتم إعادة توزيع أقسامه.

// مستهلكان في نفس المجموعة سيقرأان أقساماً مختلفةconst consumer1 = kafka.consumer({ groupId: 'order-group' });const consumer2 = kafka.consumer({ groupId: 'order-group' });

// كل منهما يشترك في نفس الموضوعawait consumer1.subscribe({ topic: 'orders' });await consumer2.subscribe({ topic: 'orders' });