当前位置：首页 > article >正文

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

article 2025/6/29 18:04:52

文章目录

从零开始掌握Kafka
- 一、Kafka 基础知识理解（理论）
- - 1. 核心组件与架构
  - 2. 重点概念解析
- 二、Kafka 面试重点知识梳理
- 三、C++ 使用 Kafka 的实践（librdkafka）
- - 1. librdkafka 简介
  - 2. 安装 librdkafka
- 四、实战：高吞吐生产者与消费者
- - 1. 生产者示例（Producer.cpp）
  - 2. 消费者示例（Consumer.cpp）
- 五、Kafka 开发相关 C++ 能力
- 六、推荐资料与开源项目

从零开始掌握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

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

此处只是简单介绍，完整工程见kafka简单工程

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

#include <librdkafka/rdkafkacpp.h>
#include <iostream>
#include <csignal>
#include <memory>class ExampleEventCb : public RdKafka::EventCb {void event_cb(RdKafka::Event &event) override {if (event.type() == RdKafka::Event::EVENT_ERROR) {std::cerr << "Kafka Error: " << event.str() << std::endl;}}
};int main() {std::string brokers = "kafka:9092";std::string topic_str = "test_topic";std::string errstr;// 配置ExampleEventCb event_cb;std::unique_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));conf->set("bootstrap.servers", brokers, errstr);conf->set("event_cb", &event_cb, errstr);// 创建 producerstd::unique_ptr<RdKafka::Producer> producer(RdKafka::Producer::create(conf.get(), errstr));if (!producer) {std::cerr << "Failed to create producer: " << errstr << std::endl;return 1;}// 创建 Topicstd::unique_ptr<RdKafka::Topic> topic(RdKafka::Topic::create(producer.get(), topic_str, nullptr, errstr));if (!topic) {std::cerr << "Failed to create topic: " << errstr << std::endl;return 1;}std::string message = "Hello from C++ Kafka Producer!";RdKafka::ErrorCode resp = producer->produce(topic.get(),                            // topic ptrRdKafka::Topic::PARTITION_UA,           // partitionRdKafka::Producer::RK_MSG_COPY,         // message flagsconst_cast<char *>(message.c_str()),    // payloadmessage.size(),                         // payload sizenullptr,                                // optional keynullptr);                               // opaqueif (resp != RdKafka::ERR_NO_ERROR) {std::cerr << "Produce failed: " << RdKafka::err2str(resp) << std::endl;} else {std::cout << "Message sent successfully\n";}producer->flush(3000);return 0;
}}

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

#include <librdkafka/rdkafkacpp.h>
#include <iostream>
#include <csignal>
#include <memory>bool running = true;void signal_handler(int) {running = false;
}class ExampleEventCb : public RdKafka::EventCb {void event_cb(RdKafka::Event &event) override {if (event.type() == RdKafka::Event::EVENT_ERROR) {std::cerr << "Kafka Error: " << event.str() << std::endl;}}
};int main() {signal(SIGINT, signal_handler);std::string brokers = "kafka:9092";std::string topic = "test_topic";std::string group_id = "cpp_consumer_group";std::string errstr;ExampleEventCb event_cb;auto conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);conf->set("bootstrap.servers", brokers, errstr);conf->set("group.id", group_id, errstr);conf->set("auto.offset.reset", "earliest", errstr);conf->set("event_cb", &event_cb, errstr);auto consumer = RdKafka::KafkaConsumer::create(conf, errstr);if (!consumer) {std::cerr << "Failed to create consumer: " << errstr << std::endl;return 1;}consumer->subscribe({topic});std::cout << "Consuming messages from topic " << topic << std::endl;while (running) {auto msg = consumer->consume(1000);if (msg->err() == RdKafka::ERR_NO_ERROR) {std::string message(reinterpret_cast<const char*>(msg->payload()), msg->len());std::cout << "Received message: " << message << std::endl;}}delete msg;}consumer->close();delete consumer;return 0;
}