العمل مع Kafka في JavaScript (Node.js): مكتبة node-rdkafka

Apache Kafka و Node.js: التعرف على node-rdkafka

Apache Kafka هي منصة موزعة لتدفق الأحداث، أصبحت المعيار الفعلي لبناء أنظمة تبادل بيانات موثوقة وقابلة للتوسع. في نظام Node.js البيئي، واحدة من أكثر المكتبات أداءً ونضجًا للعمل مع Kafka هي node-rdkafka. وهي تمثل روابط أصلية لمكتبة librdkafka (المكتوبة بلغة C/C++)، مما يضمن أقصى سرعة واستقرار.

في هذه المقالة، سنشرح بالتفصيل كيفية إعداد واستخدام node-rdkafka لإرسال (producer) وقراءة (consumer) الرسائل. ستتعلم كيفية دمج Kafka في تطبيق Node.js الخاص بك، ومعالجة الأخطاء، وإدارة تدفقات البيانات.

تثبيت node-rdkafka

لتثبيت المكتبة، يلزم وجود Node.js الإصدار 12 أو أعلى ومترجم C++ (gcc أو clang أو MSVC). يتم التثبيت باستخدام مدير الحزم القياسي:

npm install node-rdkafka

إذا فشل البناء، تأكد من تثبيت build-essential (لنظام Linux) أو Python وأدوات البناء لنظام Windows (npm install --global windows-build-tools).

إنشاء Producer (مرسل الرسائل)

الـ Producer في Kafka مسؤول عن نشر الرسائل في المواضيع. دعنا نلقي نظرة على مثال أساسي:

const Kafka = require('node-rdkafka');

// إنشاء المنتجconst producer = new Kafka.Producer({  'metadata.broker.list': 'localhost:9092',  'dr_cb': true  // رد اتصال لتأكيد التسليم});

// معالج الأحداث عند الاتصالproducer.on('event.log', function(event) {  console.log('سجل:', event.message);});

// معالج الأخطاءproducer.on('event.error', function(err) {  console.error('خطأ المنتج:', err);});

// الاتصال بالوسيطproducer.connect();

// بعد الاتصال، نرسل الرسائلproducer.on('ready', function() {  try {    // إرسال رسالة إلى الموضوع 'test-topic'    producer.produce(      'test-topic',  // اسم الموضوع      null,          // القسم (null = توزيع تلقائي)      Buffer.from('مرحباً، Kafka!'),  // نص الرسالة (Buffer)      'my-key',      // مفتاح الرسالة (اختياري)      Date.now()     // الطابع الزمني (اختياري)    );    console.log('تم إرسال الرسالة');  } catch (err) {    console.error('خطأ أثناء الإرسال:', err);  }});

// لا تنس استدعاء poll لمعالجة عمليات رد الاتصالsetInterval(() => {  producer.poll();}, 1000);

// إغلاق الاتصال عند إنهاء العمليةprocess.on('SIGINT', () => {  producer.disconnect();  process.exit();});

لاحظ: طريقة producer.produce() تستقبل نص الرسالة كـ Buffer. هذا يسمح بنقل كل من السلاسل النصية والبيانات الثنائية. المعامل dr_cb: true يفعل إشعارات رد الاتصال بالتسليم (تقرير التسليم).

إنشاء Consumer (مستقبل الرسائل)

الـ Consumer يشترك في موضوع واحد أو أكثر ويعالج الرسائل الواردة. للعمل مع مجموعات المستهلكين، يتم استخدام KafkaConsumer:

const Kafka = require('node-rdkafka');

const consumer = new Kafka.KafkaConsumer({  'group.id': 'my-group',  'metadata.broker.list': 'localhost:9092',  'enable.auto.commit': true}, {});

consumer.on('event.error', function(err) {  console.error('خطأ المستهلك:', err);});

consumer.on('ready', function() {  console.log('المستهلك جاهز');  // الاشتراك في الموضوع  consumer.subscribe(['test-topic']);  // بدء دورة الاستهلاك  consumer.consume();});

// معالجة كل رسالةconsumer.on('data', function(data) {  console.log('تم استلام الرسالة:');  console.log('  الموضوع:', data.topic);  console.log('  القسم:', data.partition);  console.log('  المفتاح:', data.key ? data.key.toString() : null);  console.log('  القيمة:', data.value.toString());