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

从零开始掌握Kafka

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

1. 核心组件与架构

组件	作用
Broker	Kafka 节点，负责存储消息
Topic	消息主题，逻辑上的分类
Partition	一个 Topic 的分片，支持并发与扩展性
Producer	负责发送消息
Consumer	负责消费消息
Consumer Group	多消费者协作消费
Zookeeper / KRaft	负责元数据与协调（未来版本转向 KRaft 模式）

2. 重点概念解析

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

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

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

• offset 管理：

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

---

二、Kafka 面试重点知识梳理

面试点	说明
消息顺序性	同一个 partition 内有顺序，跨 partition 无法保证
幂等性生产	使用 enable.idempotence=true，避免 producer 重试造成重复发送
分布式一致性	ISR 机制，消息写入需同步到 follower；ACK=all 实现强一致
消费位点提交	手动提交 offset 是保证消费语义精确一次的关键
Rebalance 原理	消费者上下线会触发 Rebalance，导致 partition 分配变化

---

三、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. 模块设计建议

模块	功能
KafkaProducer	封装 Producer 初始化/发送逻辑
KafkaConsumer	封装 Consumer + offset 管理
ConfigManager	读取配置
Message	消息封装与解析（支持 JSON）

---

七、推荐资料与开源项目

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

• 📘 librdkafka 文档

• 📖 Apache Kafka 官方文档

• 💻 开源项目参考：

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