TypeScript में Kafka का उपयोग: Node.js के लिए KafkaJS लाइब्रेरी

KafkaJS लाइब्रेरी: TypeScript में Apache Kafka के साथ काम करना

Apache Kafka एक डिस्ट्रीब्यूटेड डेटा स्ट्रीमिंग प्लेटफ़ॉर्म है, जिसका उपयोग हाई-लोड मैसेजिंग सिस्टम बनाने के लिए किया जाता है। TypeScript (और Node.js) इकोसिस्टम में KafkaJS लाइब्रेरी डी-फैक्टो स्टैंडर्ड बन गई है। यह एक क्लीन, मॉडर्न API प्रदान करती है जिसमें TypeScript का पूरा सपोर्ट है, जिससे डेवलपर्स Kafka प्रोड्यूसर और कंज्यूमर के लिए टाइप-सेफ कोड लिख सकते हैं।

इस लेख में हम KafkaJS की इंस्टॉलेशन, कॉन्फ़िगरेशन और मुख्य उपयोग के तरीकों को विस्तार से समझेंगे। आप सीखेंगे कि मैसेज कैसे भेजें और प्राप्त करें, एरर को कैसे हैंडल करें और प्रोडक्शन एनवायरनमेंट के लिए क्लाइंट को सही तरीके से कॉन्फ़िगर करें।

KafkaJS की इंस्टॉलेशन और सेटअप

शुरू करने के लिए, npm या yarn के माध्यम से लाइब्रेरी इंस्टॉल करें:

npm install kafkajs
# या
yarn add kafkajs

KafkaJS TypeScript में लिखी गई है, इसलिए टाइप्स पहले से बिल्ट-इन हैं — @types/kafkajs अलग से इंस्टॉल करने की ज़रूरत नहीं है।

अब Kafka क्लाइंट का एक इंस्टेंस बनाते हैं। फॉल्ट-टॉलरेंस के लिए कई ब्रोकर्स निर्दिष्ट करना बेहतर होता है:

import { Kafka, logLevel } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092', 'localhost:9093'],
  logLevel: logLevel.INFO, // लॉगिंग लेवल
  retry: {
    initialRetryTime: 100,
    retries: 8
  }
});

प्रोड्यूसर: Kafka में मैसेज भेजना

प्रोड्यूसर Kafka टॉपिक्स में मैसेज लिखने के लिए जिम्मेदार होता है। KafkaJS अलग-अलग मैसेज और बैच दोनों भेजने के लिए एक आसान API प्रदान करता है।

बेसिक प्रोड्यूसर का उदाहरण:

import { Kafka, Producer } from 'kafkajs';

async function produceMessage(producer: Producer) {
  await producer.send({
    topic: 'orders',
    messages: [
      { 
        key: 'order-123', 
        value: JSON.stringify({ userId: 42, amount: 1500 }) 
      },
    ],
  });
  console.log('मैसेज भेज दिया गया');
}

async function runProducer() {
  const producer = kafka.producer();
  await producer.connect();

  // कई मैसेज भेजना
  await producer.send({
    topic: 'orders',
    messages: [
      { key: 'order-124', value: '{"item":"laptop"}' },
      { key: 'order-125', value: '{"item":"mouse"}' },
    ],
  });

  await produceMessage(producer);
  await producer.disconnect();
}

runProducer().catch(console.error);

महत्वपूर्ण: काम खत्म होने के बाद हमेशा disconnect() कॉल करें ताकि संसाधन खाली हो जाएं। लंबे समय तक चलने वाले ऐप्स के लिए प्रोड्यूसर को लगातार कनेक्टेड रखा जा सकता है।

कंज्यूमर: Kafka से मैसेज पढ़ना

कंज्यूमर एक या अधिक टॉपिक्स को सब्सक्राइब करता है और मैसेज प्रोसेस करता है। KafkaJS समानांतर प्रोसेसिंग के लिए कंज्यूमर ग्रुप्स को सपोर्ट करता है।

मैसेज प्रोसेसिंग के साथ कंज्यूमर का उदाहरण:

import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';

async function runConsumer() {
  const consumer: Consumer = kafka.consumer({ 
    groupId: 'order-processor-group' 
  });

  await consumer.connect();
  await consumer.subscribe({ 
    topic: 'orders', 
    fromBeginning: true // पहले मैसेज से पढ़ना शुरू करें
  });

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

runConsumer().catch(console.error);

fromBeginning पैरामीटर पर ध्यान दें: अगर true सेट किया जाए, तो कंज्यूमर पहले से मौजूद सभी मैसेज पढ़ लेगा। प्रोडक्शन एनवायरनमेंट में आमतौर पर false (केवल नए मैसेज) का उपयोग किया जाता है।

एरर हैंडलिंग और रीट्राई

रियल-वर्ल्ड एप्लिकेशन में एरर को सही तरीके से हैंडल करना ज़रूरी है।