Apache Kafka C++ में उपयोग: लाइब्रेरी और उदाहरण

C++ में Apache Kafka के साथ काम: लाइब्रेरी, सेटअप और उदाहरण

Apache Kafka एक डिस्ट्रीब्यूटेड डेटा स्ट्रीमिंग प्लेटफॉर्म है, जिसका उपयोग माइक्रोसर्विसेज, लॉगिंग सिस्टम और एनालिटिक्स में बड़े पैमाने पर किया जाता है। C++ के साथ Kafka को इंटीग्रेट करने के लिए कई परिपक्व लाइब्रेरी उपलब्ध हैं, जिनमें से सबसे लोकप्रिय हैं librdkafka (लो-लेवल C लाइब्रेरी) और cppkafka (C++ रैपर)। इस लेख में हम दोनों लाइब्रेरीज को विस्तार से समझेंगे, प्रोड्यूसर और कंज्यूमर के कोड उदाहरण दिखाएंगे, और चुनने के लिए सुझाव देंगे।

1. librdkafka — C में Kafka के लिए बुनियादी लाइब्रेरी

librdkafka Confluent द्वारा विकसित एक हाई-परफॉर्मेंस C लाइब्रेरी है, जो Apache Kafka प्रोटोकॉल को लागू करती है। यह सभी प्रमुख फीचर्स को सपोर्ट करती है: एसिंक्रोनस मैसेज सेंडिंग, कंज्यूमर ग्रुप्स, exactly-once सिमैंटिक्स (ट्रांजैक्शन के जरिए) और SSL/SASL। यह लाइब्रेरी C में लिखी गई है, लेकिन इसे C++ से सीधे या C++ रैपर के जरिए इस्तेमाल किया जा सकता है।

librdkafka इंस्टॉल करना

Ubuntu/Debian पर:

sudo apt-get install librdkafka-dev

macOS पर Homebrew के जरिए:

brew install librdkafka

सोर्स कोड से बिल्ड करना:

git clone https://github.com/confluentinc/librdkafka.git
cd librdkafka
./configure --prefix=/usr/local
make
sudo make install

librdkafka के साथ C प्रोड्यूसर का उदाहरण

एक सरल प्रोड्यूसर जो टॉपिक में मैसेज भेजता है:

#include <librdkafka/rdkafka.h>
#include <iostream>
#include <string>
#include <cstring>

int main() {
    rd_kafka_t *rk;          // प्रोड्यूसर हैंडलर
    rd_kafka_conf_t *conf;   // कॉन्फ़िगरेशन
    char errstr[512];

    conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "bootstrap.servers", "localhost:9092", errstr, sizeof(errstr));
    rd_kafka_conf_set(conf, "acks", "all", errstr, sizeof(errstr));

    rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
    if (!rk) {
        std::cerr << "प्रोड्यूसर बनाने में विफल: " << errstr << std::endl;
        return 1;
    }

    std::string topic_name = "test-topic";
    std::string message = "C++ और librdkafka से नमस्ते!";

    rd_kafka_resp_err_t err = rd_kafka_producev(
        rk,
        RD_KAFKA_V_TOPIC(topic_name.c_str()),
        RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
        RD_KAFKA_V_VALUE((void*)message.c_str(), message.size()),
        RD_KAFKA_V_END
    );

    if (err) {
        std::cerr << "मैसेज भेजने में विफल: " << rd_kafka_err2str(err) << std::endl;
    } else {
        std::cout << "मैसेज सफलतापूर्वक भेजा गया!" << std::endl;
    }

    // डिलीवरी का इंतजार और सफाई
    rd_kafka_flush(rk, 10000);
    rd_kafka_destroy(rk);
    return 0;
}

librdkafka के साथ C कंज्यूमर का उदाहरण

एक सरल कंज्यूमर जो टॉपिक से मैसेज पढ़ता है:

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

int main() {
    rd_kafka_t *rk;
    rd_kafka_conf_t *conf;
    rd_kafka_topic_partition_list_t *topics;
    char errstr[512];

    conf = rd_kafka_conf_new();
    rd_kafka_conf_set(conf, "bootstrap.servers", "localhost:9092", errstr, sizeof(errstr));
    rd_kafka_conf_set(conf, "group.id", "my-group", errstr, sizeof(errstr));
    rd_kafka_conf_set(conf, "auto.offset.reset", "earliest", errstr, sizeof(errstr));

    rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));
    if (!rk) {
        std::cerr << "कंज्यूमर बनाने में विफल: " << errstr << std::endl;
        return 1;
    }

    rd_kafka_poll_set_consumer(rk);
    topics = rd_kafka_topic_partition_list_new(1);
    rd_kafka_topic_partition_list_add(topics, "test-topic", RD_KAFKA_PARTITION_UA);
    rd_kafka_consumer_subscribe(rk, topics);

    std::cout << "मैसेज का इंतजार कर रहे हैं..." << std::endl;
    while (true) {
        rd_kafka_message_t *msg = rd_kafka_consumer_poll(rk,