在 C++ 中使用 Apache Kafka：librdkafka 库及实践示例

librdkafka 库：在 C++ 中使用 Kafka 的强大工具

Apache Kafka 是一个分布式事件流平台，已成为构建高负载消息交换系统的事实标准。要将 Kafka 与 C++ 集成，最流行且成熟的库是 librdkafka——一个高性能的 C 语言库，提供了与 Kafka 生产者和消费者协作的完整功能集。它实现了 Apache Kafka 0.8 及以上版本的协议，包括最新的稳定版本。

librdkafka 的主要功能

• 支持所有关键操作：发送消息（生产者）和读取消息（消费者），并支持偏移量管理。
• 支持主题、分区和消费者组。
• 原生支持压缩（gzip、snappy、lz4、zstd）。
• SSL/TLS 加密和 SASL 认证（PLAIN、SCRAM、GSSAPI、OAUTHBEARER）。
• 幂等生产者和事务（从 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

示例：创建生产者（Producer）

生产者将消息发送到指定主题。以下是一个最小的工作示例：

#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 方法确保在程序结束前所有消息都已发送。

示例：创建消费者（Consumer）

消费者作为消费者组的一部分从主题中读取消息。以下是手动管理偏移量的示例：

#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