PHP में Kafka का उपयोग: लाइब्रेरी, कनेक्शन और कोड उदाहरण

PHP में Kafka का उपयोग: लाइब्रेरी, कनेक्शन और कोड उदाहरण

Apache Kafka एक डिस्ट्रीब्यूटेड डेटा स्ट्रीमिंग प्लेटफॉर्म है जो रियल-टाइम में मैसेज को पब्लिश, स्टोर और प्रोसेस करने की सुविधा देता है। PHP इकोसिस्टम में Kafka के साथ काम करने के लिए php-rdkafka एक्सटेंशन का उपयोग किया जाता है - यह C में लिखी गई librdkafka लाइब्रेरी का रैपर है। इस लेख में हम जानेंगे कि इसे कैसे इंस्टॉल करें, प्रोड्यूसर और कंज्यूमर कैसे लिखें, और इसके सामान्य उपयोग के मामले क्या हैं।

php-rdkafka क्या है?

php-rdkafka Apache Kafka के साथ काम करने के लिए आधिकारिक PHP एक्सटेंशन है। यह मैसेज भेजने और प्राप्त करने के लिए हाई-परफॉर्मेंस इंटरफेस प्रदान करता है। यह एक्सटेंशन Kafka की सभी प्रमुख सुविधाओं को सपोर्ट करता है: टॉपिक, पार्टीशन, कंज्यूमर ग्रुप, मैन्युअल कमिट, कंप्रेशन और बहुत कुछ।

मुख्य विशेषताएं:

• टॉपिक में मैसेज भेजने के लिए प्रोड्यूसर (producers) बनाना।
• टॉपिक से मैसेज पढ़ने के लिए कंज्यूमर (consumers) बनाना।
• कंज्यूमर ग्रुप और ऑटोमैटिक रीबैलेंसिंग का सपोर्ट।
• पार्टीशन के साथ काम करना और मैन्युअल offset प्रबंधन।
• मैसेज कंप्रेशन (gzip, snappy, lz4, zstd)।
• एसिंक्रोनस सेंडिंग और बैच प्रोसेसिंग।

php-rdkafka इंस्टॉलेशन

एक्सटेंशन को PECL के माध्यम से या सोर्स कोड से बिल्ड करके इंस्टॉल किया जा सकता है। अनुशंसित तरीका PECL के माध्यम से है:

# PECL के माध्यम से इंस्टॉल करें
pecl install rdkafka

# Ubuntu/Debian के लिए पहले librdkafka-dev इंस्टॉल करें
sudo apt-get install librdkafka-dev
pecl install rdkafka

# php.ini में एक्सटेंशन जोड़ें
extension=rdkafka.so

इंस्टॉलेशन के बाद जांचें कि एक्सटेंशन लोड हुआ है या नहीं:

php -m | grep rdkafka

यदि आप Composer का उपयोग कर रहे हैं, तो ऑटोलोडिंग के लिए arnaud-lb/php-rdkafka पैकेज भी इंस्टॉल कर सकते हैं:

composer require arnaud-lb/php-rdkafka

प्रोड्यूसर (Producer) का उदाहरण

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

<?php
require_once 'vendor/autoload.php';

$conf = new RdKafka\Conf();
$conf->set('metadata.broker.list', 'localhost:9092');

$producer = new RdKafka\Producer($conf);
$topic = $producer->newTopic('test_topic');

// मैसेज भेजना
$topic->produce(RD_KAFKA_PARTITION_UA, 0, 'Hello, Kafka from PHP!');

// सभी मैसेज भेजने की प्रतीक्षा करें
$producer->flush(1000);
echo "मैसेज भेज दिया गया\n";

कोड की व्याख्या:

• metadata.broker.list — Kafka ब्रोकर का पता (डिफ़ॉल्ट localhost:9092)।
• RD_KAFKA_PARTITION_UA — ऑटोमैटिक पार्टीशन चयन।
• flush(1000) — सभी मैसेज भेजने तक एक्जीक्यूशन को ब्लॉक करता है (टाइमआउट 1 सेकंड)।

कंज्यूमर (Consumer) का उदाहरण

कंज्यूमर टॉपिक से मैसेज पढ़ता है। कंज्यूमर ग्रुप और offset कमिट स्ट्रेटेजी निर्दिष्ट करना महत्वपूर्ण है:

<?php
require_once 'vendor/autoload.php';

$conf = new RdKafka\Conf();
$conf->set('metadata.broker.list', 'localhost:9092');
$conf->set('group.id', 'my_consumer_group');
$conf->set('auto.offset.reset', 'earliest');

$consumer = new RdKafka\KafkaConsumer($conf);
$consumer->subscribe(['test_topic']);

echo "मैसेज की प्रतीक्षा...\n";

while (true) {
    $message = $consumer->consume(120 * 1000); // टाइमआउट 2 मिनट
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            echo "मैसेज प्राप्त हुआ: " . $message->payload . "\n";
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "पार्टीशन का अंत\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "टाइमआउट\n";
            break;
        default:
            echo "त्रुटि: " . $message->errstr() . "\n";
            break;
    }
}