【C/C++】从零开始掌握Kafka

从零开始掌握Kafka

一、Kafka 基础知识理解（理论）

1. 核心组件与架构

2. 重点概念解析

• Partition：分片，支持水平扩展（每个 partition 是一个有序日志）。

• 副本机制（Replication）：每个 Partition 有一个 leader + N 个 follower，保证高可用。

• 消费者组（Consumer Group）：Kafka 实现广播和负载均衡消费的机制。

• offset 管理：

  • 自动提交（enable.auto.commit）
  • 手动提交（commitSync / commitAsync）
  • Kafka 默认 offset 存在 __consumer_offsets topic 中。

二、Kafka 面试重点知识梳理

三、C++ 使用 Kafka 的实践（librdkafka）

1. librdkafka 简介

• 官方提供的高性能 C/C++ Kafka 客户端库。

• GitHub 地址：https://github.com/edenhill/librdkafka

• 支持：

  • 高吞吐的生产与消费
  • offset 提交
  • topic/partition 管理
  • 幂等发送、压缩、批处理

2. 安装 librdkafka

# Ubuntu
sudo apt-get install librdkafka-dev

# Or from source
git clone https://github.com/edenhill/librdkafka.git
cd librdkafka
./configure
make
sudo make install

四、实战：高吞吐生产者与消费者

1. 生产者示例（Producer.cpp）

#include <librdkafka/rdkafkacpp.h>

class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message &message) override {
        if (message.err()) {
            std::cerr << "Delivery failed: " << message.errstr() << std::endl;
        } else {
            std::cout << "Delivered message to " << message.topic_name()
                      << " [" << message.partition() << "] at offset "
                      << message.offset() << std::endl;
        }
    }
};

int main() {
    std::string errstr;
    std::string topic = "test_topic";

    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    conf->set("bootstrap.servers", "localhost:9092", errstr);

    ExampleDeliveryReportCb dr_cb;
    conf->set("dr_cb", &dr_cb, errstr);

    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
    delete conf;

    for (int i = 0; i < 10000; ++i) {
        std::string message = "Message " + std::to_string(i);
        producer->produce(topic, RdKafka::Topic::PARTITION_UA,
                          RdKafka::Producer::RK_MSG_COPY,
                          const_cast<char *>(message.c_str()), message.size(),
                          nullptr, 0, 0, nullptr);
        producer->poll(0); // 异步发送需要poll触发回调
    }

    producer->flush(10000); // 等待所有消息发送完成
    delete producer;
}

2. 消费者示例（Consumer.cpp）

#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);
    conf->set("group.id", "my_group", errstr);
    conf->set("enable.auto.commit", "false", errstr); // 手动提交

    RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);
    delete conf;

    std::vector<std::string> topics = {"test_topic"};
    consumer->subscribe(topics);

    while (true) {
        RdKafka::Message *msg = consumer->consume(1000);
        switch (msg->err()) {
            case RdKafka::ERR_NO_ERROR:
                std::cout << "Read msg at offset " << msg->offset()
                          << ": " << static_cast<const char *>(msg->payload()) << std::endl;
                consumer->commitAsync(msg);  // 或 commitSync(msg)
                break;
            case RdKafka::ERR__TIMED_OUT:
                break;
            default:
                std::cerr << "Error: " << msg->errstr() << std::endl;
        }
        delete msg;
    }

    consumer->close();
    delete consumer;
}

五、Kafka 开发相关 C++ 能力要求

必备 C++ 能力

• 熟练使用 RAII、智能指针、异常处理
• 理解线程安全、异步模型（poll, callback）
• 能够结合 JSON/XML 配置 Kafka 客户端
• 编写模块化、高性能的消息收发组件

推荐工具链

• 构建系统：CMake
• 日志：spdlog 或 glog
• 单元测试：gtest
• JSON：nlohmann/json

六、项目实践建议

1. 项目目标

构建一个 C++ Kafka 模块，支持：

• 高吞吐量生产（批量发送，压缩）
• 幂等性保障
• 多线程消费 + 位点提交策略
• 简单的失败重试机制
• 支持 JSON 格式的序列化消息

2. 模块设计建议

七、推荐资料与开源项目

• 📚 Kafka 权威指南（原书第2版）

• 📘 librdkafka 文档

• 📖 Apache Kafka 官方文档

• 💻 开源项目参考：

  • confluent-kafka-cpp
  • cppkafka（封装更现代 C++）