rdkafka详解

一、rdkafka简介

rdkafka是一个高性能的消息队列中间件，它基于C++编写而成，具有很好的可扩展性和可靠性，适用于各类分布式场景下的数据传输等场景，目前已应用于多个大型在线系统中。

rdkafka主要特点如下：

1、高吞吐量：消息传输快速

2、支持多语言：除了C++，还支持Java、Python等语言

3、支持多种传输协议：比如TCP、UDP等协议

4、高可靠性：消息传输过程中保证幂等性和事务性

5、易扩展：支持集群和分布式部署

二、rdkafka使用方法

1. 安装rdkafka

在Linux系统下，可以通过以下命令安装rdkafka：

git clone https://github.com/edenhill/librdkafka.git
cd librdkafka
./configure
make
make install

2. rdafka C++ API的使用

2.1 生产者示例

以下是rdkafka生产者示例的代码：

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

int main() {
    std::string errstr;
    RdKafka::Producer *producer = RdKafka::Producer::create({
        {"bootstrap.servers", "localhost"},
        {"message.timeout.ms", "3000"}
    }, errstr);

    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        return 1;
    }

    RdKafka::Topic *topic = RdKafka::Topic::create(producer, "test", nullptr, errstr);

    if (!topic) {
        std::cerr << "Failed to create topic: " << errstr << std::endl;
        return 1;
    }

    // 创建消息
    std::string payload = "hello world";
    RdKafka::Producer::Message* message = RdKafka::Producer::Message::create(topic,
            RdKafka::Producer::RK_MSG_COPY /*复制消息体内容*/,
            const_cast<char*>(payload.c_str()), payload.size(), nullptr, nullptr);

    producer->produce(topic, RdKafka::Topic::PARTITION_UA /*指定分区，这里使用默认值*/, RdKafka::Producer::RK_MSG_COPY /*复制消息体内容*/, message, nullptr);

    // 同步等待生产者发送消息的结果
    producer->flush(RdKafka::Topic::PARTITION_UA, 3000);
}

2.2 消费者示例

以下是rdkafka消费者示例的代码：

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

class MyEventCb : public RdKafka::EventCb {
public:
    void event_cb (RdKafka::Event &event) {
        switch (event.type())
        {
        case RdKafka::Event::EVENT_ERROR:
            if (event.err()) {
                std::cerr << "ERROR (event): " << RdKafka::err2str(event.err()) << std::endl;
            }
            break;
        case RdKafka::Event::EVENT_LOG:
            std::cout << "LOG-" << event.severity() << "-" << event.fac() << ": " << event.str() << std::endl;
            break;
        default:
            std::cerr << "Unhandled event type: " << event.type() << std::endl;
            break;
        }
    }
};

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

    // 配置消费者参数
    if (conf->set("group.id", "test", errstr) != RdKafka::Conf::CONF_OK) {
        std::cerr << errstr << std::endl;
        return 1;
    }

    // 设置回调函数
    MyEventCb eventCb;
    conf->set("event_cb", &eventCb, errstr);

    // 创建消费者实例
    RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);

    if (!consumer) {
        std::cerr << "Failed to create consumer: " << errstr << std::endl;
        return 1;
    }

    // 订阅主题
    std::vector<std::string> topics = {"test"};
    if (consumer->subscribe(topics) != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Failed to subscribe to topics: " << errstr << std::endl;
        return 1;
    }

    while(true) {
        // 接收消息
        RdKafka::Message *message = consumer->consume(1000);

        if (!message) {
            continue;
        }

        if (message->err() == RdKafka::ERR_NO_ERROR) {
            std::cout << "Received message:" << std::endl;
            std::cout << "payload:" << std::string(static_cast<const char*>(message->payload()), message->len()) << std::endl;
        }

        delete message;
    }

    return 0;
}

三、rdkafka集成到项目中的示例

以下是rdkafka集成到项目中的示例代码：

#include <librdkafka/rdkafkacpp.h>
#include <iostream>
#include <thread>
#include <chrono>

class KafkaConsumer {
public:
    KafkaConsumer() {
        conf_ = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
        std::string errstr;

        // 指定Kafka集群地址
        if (conf_->set("bootstrap.servers", "localhost:9092", errstr) != RdKafka::Conf::CONF_OK) {
            throw std::runtime_error("Failed to set broker configuration: " + errstr);
        }

        // 配置消费者ID
        if (conf_->set("group.id", "test-group", errstr) != RdKafka::Conf::CONF_OK) {
            throw std::runtime_error("Failed to set group.id configuration: " + errstr);
        }

        // 创建消费者实例
        consumer_ = RdKafka::KafkaConsumer::create(conf_.get(), errstr);

        if (!consumer_) {
            throw std::runtime_error("Failed to create KafkaConsumer: " + errstr);
        }

        // 订阅主题
        std::vector<std::string> topics = {"test-topic"};
        if (consumer_->subscribe(topics) != RdKafka::ERR_NO_ERROR) {
            throw std::runtime_error("Failed to subscribe to topics: " + errstr);
        }
    }

    ~KafkaConsumer() {
        consumer_->close();
        delete consumer_;
    }

    void consumeMessage() {
        while (true) {
            // 从Kafka服务器接收消息
            RdKafka::Message *message = consumer_->consume(1000);

            if (!message) {
                continue;
            }

            if (message->err() == RdKafka::ERR_NO_ERROR) {
                std::cout << "Received message: "
                        << std::string(static_cast<const char*>(message->payload()), message->len()) << std::endl;
            }

            delete message;
        }
    }

private:
    std::unique_ptr<RdKafka::Conf> conf_;
    RdKafka::KafkaConsumer *consumer_;
};

class KafkaProducer {
public:
    KafkaProducer() {
        conf_ = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
        std::string errstr;

        // 指定Kafka集群地址
        if (conf_->set("bootstrap.servers", "localhost:9092", errstr) != RdKafka::Conf::CONF_OK) {
            throw std::runtime_error("Failed to set broker configuration: " + errstr);
        }

        // 创建生产者实例
        producer_ = RdKafka::Producer::create(conf_.get(), errstr);

        if (!producer_) {
            throw std::runtime_error("Failed to create KafkaProducer: " + errstr);
        }
    }

    ~KafkaProducer() {
        producer_->flush(0);
        delete producer_;
    }

    void produceMessage(const std::string& message) {
        RdKafka::Topic *topic = RdKafka::Topic::create(producer_, "test-topic", nullptr, errstr_);

        if (!topic) {
            throw std::runtime_error("Failed to create topic: " + errstr_);
        }

        // 创建消息
        RdKafka::Producer::Message* msg = RdKafka::Producer::Message::create(
                topic,
                RdKafka::Producer::RK_MSG_COPY /*复制消息体内容*/,
                const_cast<char*>(message.c_str()), message.size(), nullptr, nullptr);

        // 发送消息，不需要等待ack
        producer_->produce(topic, RdKafka::Topic::PARTITION_UA /*指定分区，这里使用默认值*/, RdKafka::Producer::RK_MSG_COPY /*复制消息体内容*/, msg, nullptr);
        delete msg;
    }

private:
    std::unique_ptr<RdKafka::Conf> conf_;
    RdKafka::Producer *producer_;
    std::string errstr_;
};

int main() {
    KafkaProducer producer;
    producer.produceMessage("Hello World!");

    // 延迟1s，确保消息已经被消费
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    KafkaConsumer consumer;
    consumer.consumeMessage();
    return 0;
}