JavaScript (Node.js) में Kafka के साथ कार्य: node-rdkafka लाइब्रेरी

Apache Kafka और Node.js: node-rdkafka से परिचय

Apache Kafka एक वितरित इवेंट स्ट्रीमिंग प्लेटफ़ॉर्म है, जो विश्वसनीय और स्केलेबल डेटा एक्सचेंज सिस्टम बनाने के लिए डी-फ़ैक्टो मानक बन गया है। Node.js इकोसिस्टम में, Kafka के साथ काम करने के लिए सबसे अधिक प्रदर्शनशील और परिपक्व लाइब्रेरीज़ में से एक node-rdkafka है। यह librdkafka लाइब्रेरी (C/C++ में लिखी गई) के लिए नेटिव बाइंडिंग प्रदान करता है, जो अधिकतम गति और स्थिरता सुनिश्चित करता है।

इस लेख में, हम विस्तार से समझेंगे कि संदेश भेजने (producer) और पढ़ने (consumer) के लिए node-rdkafka को कैसे सेट अप और उपयोग करें। आप सीखेंगे कि 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 (संदेश भेजने वाला) बनाना

Kafka में Producer टॉपिक में संदेश प्रकाशित करने के लिए जिम्मेदार होता है। आइए एक बुनियादी उदाहरण देखें:

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

// प्रोड्यूसर बनाएँconst producer = new Kafka.Producer({  'metadata.broker.list': 'localhost:9092',  'dr_cb': true  // डिलीवरी पुष्टिकरण के लिए Callback});

// कनेक्ट होने पर इवेंट हैंडलर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);  }});

// Callback प्रोसेस करने के लिए poll को कॉल करना न भूलेंsetInterval(() => {  producer.poll();}, 1000);

// प्रक्रिया समाप्त होने पर कनेक्शन बंद करेंprocess.on('SIGINT', () => {  producer.disconnect();  process.exit();});

ध्यान दें: producer.produce() विधि संदेश के बॉडी को Buffer के रूप में स्वीकार करती है। यह स्ट्रिंग और बाइनरी डेटा दोनों को भेजने की अनुमति देता है। पैरामीटर dr_cb: true डिलीवरी रिपोर्ट (delivery report) के लिए callback सूचनाएँ सक्षम करता है।

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());