C++ में Apache Kafka के साथ कार्य: librdkafka लाइब्रेरी और व्यावहारिक उदाहरण

librdkafka लाइब्रेरी: C++ में Kafka के साथ कार्य करने का शक्तिशाली उपकरण

Apache Kafka एक वितरित इवेंट स्ट्रीमिंग प्लेटफॉर्म है, जो उच्च-लोड संदेश विनिमय प्रणालियों के निर्माण के लिए डी-फैक्टो मानक बन गया है। C++ के साथ Kafka को एकीकृत करने के लिए सबसे लोकप्रिय और परिपक्व लाइब्रेरी librdkafka है - एक उच्च-प्रदर्शन C-लाइब्रेरी, जो Kafka प्रोड्यूसर और कंज्यूमर के साथ कार्य करने के लिए पूर्ण क्षमताएं प्रदान करती है। यह Apache Kafka प्रोटोकॉल संस्करण 0.8 और उससे ऊपर, नवीनतम स्थिर संस्करणों सहित, को लागू करती है।

librdkafka की मुख्य क्षमताएं

• सभी प्रमुख संचालनों का समर्थन: संदेश भेजना (producer) और ऑफसेट प्रबंधन के साथ पढ़ना (consumer)।
• टॉपिक, पार्टीशन और उपभोक्ता समूहों के साथ कार्य।
• संपीड़न का मूल समर्थन (gzip, snappy, lz4, zstd)।
• SSL/TLS एन्क्रिप्शन और SASL प्रमाणीकरण (PLAIN, SCRAM, GSSAPI, OAUTHBEARER)।
• Idempotent producer और ट्रांजेक्शन (Kafka 0.11 से शुरू)।
• कॉलबैक और कम विलंबता के साथ अतुल्यकालिक प्रसंस्करण।
• कॉन्फ़िगरेशन गुणों के माध्यम से सूक्ष्म सेटिंग (Java क्लाइंट के समान)।

स्थापना और कनेक्शन

librdkafka को पैकेज मैनेजर के माध्यम से स्थापित किया जा सकता है या स्रोत से संकलित किया जा सकता है। Linux (Ubuntu/Debian) के लिए कमांड:

sudo apt-get install librdkafka-dev

macOS के लिए Homebrew के माध्यम से:

brew install librdkafka

स्थापना के बाद, संकलन करते समय हेडर फ़ाइल और लाइब्रेरी को शामिल करें:

g++ -o my_app my_app.cpp -lrdkafka -lpthread -lz

उदाहरण: प्रोड्यूसर बनाना

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

#include <iostream>#include <string>#include <librdkafka/rdkafkacpp.h>

int main() {    std::string errstr;    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

    // ब्रोकर सेटिंग    conf->set("bootstrap.servers", "localhost:9092", errstr);

    // प्रोड्यूसर बनाना    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);    if (!producer) {        std::cerr << "Failed to create producer: " << errstr << std::endl;        return 1;    }    delete conf;

    // संदेश भेजना    std::string topic_name = "test-topic";    std::string payload = "Hello from C++ Kafka!";    RdKafka::ErrorCode resp = producer->produce(        topic_name,        RdKafka::Topic::PARTITION_UA,        RdKafka::Producer::RK_MSG_COPY,        const_cast<char*>(payload.c_str()),        payload.size(),        nullptr, 0, 0, nullptr    );

    if (resp != RdKafka::ERR_NO_ERROR) {        std::cerr << "Produce failed: " << RdKafka::err2str(resp) << std::endl;    } else {        std::cout << "Message sent successfully" << std::endl;    }

    // डिलीवरी की प्रतीक्षा    producer->flush(10000);    delete producer;    return 0;}

इस उदाहरण में, हम एक कॉन्फ़िगरेशन बनाते हैं, ब्रोकर का पता निर्दिष्ट करते हैं, Producer का एक इंस्टेंस बनाते हैं और test-topic टॉपिक में एक स्ट्रिंग संदेश भेजते हैं। flush विधि यह सुनिश्चित करती है कि प्रोग्राम समाप्त होने से पहले सभी संदेश भेज दिए जाएं।

उदाहरण: कंज्यूमर बनाना

कंज्यूमर उपभोक्ता समूह के हिस्से के रूप में टॉपिक से संदेश पढ़ता है। मैन्युअल ऑफसेट प्रबंधन के साथ उदाहरण:

#include <iostream>#include <string>#include <librdkafka/rdkafkacpp.h>

class MyConsumerCb : public RdKafka::ConsumeCb {public:    void consume_cb(RdKafka::Message &msg, void *opaque) override {        if (msg.err()) {            if (msg.err() != RdKafka::ERR__TIMED_OUT) {                std::cerr << "Error: " << msg.errs