【C/C++】从零开始掌握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
<p># Ubuntu</p><p>sudo apt-get install librdkafka-dev</p><p>
</p><p># Or from source</p><p>git clone https://github.com/edenhill/librdkafka.git</p><p>cd librdkafka</p><p>./configure</p><p>make</p><p>sudo make install</p>
四、实战：高吞吐生产者与消费者
此处只是简单介绍，完整工程见kafka简单工程
1. 生产者示例（Producer.cpp）
<p>#include <librdkafka/rdkafkacpp.h></p><p>#include <iostream></p><p>#include <csignal></p><p>#include <memory></p><p>
</p><p>class ExampleEventCb : public RdKafka::EventCb {</p><p>void event_cb(RdKafka::Event &event) override {</p><p> if (event.type() == RdKafka::Event::EVENT_ERROR) {</p><p> std::cerr << "Kafka Error: " << event.str() << std::endl;</p><p> }</p><p>}</p><p>};</p><p>
</p><p>int main() {</p><p>std::string brokers = "kafka:9092";</p><p>std::string topic_str = "test_topic";</p><p>std::string errstr;</p><p>
</p><p>// 配置</p><p>ExampleEventCb event_cb;</p><p>std::unique_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));</p><p>conf->set("bootstrap.servers", brokers, errstr);</p><p>conf->set("event_cb", &event_cb, errstr);</p><p>
</p><p>// 创建 producer</p><p>std::unique_ptr<RdKafka::Producer> producer(RdKafka::Producer::create(conf.get(), errstr));</p><p>if (!producer) {</p><p> std::cerr << "Failed to create producer: " << errstr << std::endl;</p><p> return 1;</p><p>}</p><p>
</p><p>// 创建 Topic</p><p>std::unique_ptr<RdKafka::Topic> topic(RdKafka::Topic::create(producer.get(), topic_str, nullptr, errstr));</p><p>if (!topic) {</p><p> std::cerr << "Failed to create topic: " << errstr << std::endl;</p><p> return 1;</p><p>}</p><p>
</p><p>std::string message = "Hello from C++ Kafka Producer!";</p><p>RdKafka::ErrorCode resp = producer->produce(</p><p> topic.get(), // topic ptr</p><p> RdKafka::Topic::PARTITION_UA, // partition</p><p> RdKafka::Producer::RK_MSG_COPY, // message flags</p><p> const_cast<char *>(message.c_str()), // payload</p><p> message.size(), // payload size</p><p> nullptr, // optional key</p><p> nullptr); // opaque</p><p>
</p><p>if (resp != RdKafka::ERR_NO_ERROR) {</p><p> std::cerr << "Produce failed: " << RdKafka::err2str(resp) << std::endl;</p><p>} else {</p><p> std::cout << "Message sent successfully\n";</p><p>}</p><p>
</p><p>producer->flush(3000);</p><p>return 0;</p><p>}</p><p>
</p><p>
</p><p>}</p>
2. 消费者示例（Consumer.cpp）
<p>#include <librdkafka/rdkafkacpp.h></p><p>#include <iostream></p><p>#include <csignal></p><p>#include <memory></p><p>
</p><p>bool running = true;</p><p>
</p><p>void signal_handler(int) {</p><p>running = false;</p><p>}</p><p>
</p><p>class ExampleEventCb : public RdKafka::EventCb {</p><p>void event_cb(RdKafka::Event &event) override {</p><p> if (event.type() == RdKafka::Event::EVENT_ERROR) {</p><p> std::cerr << "Kafka Error: " << event.str() << std::endl;</p><p> }</p><p>}</p><p>};</p><p>
</p><p>int main() {</p><p>signal(SIGINT, signal_handler);</p><p>
</p><p>std::string brokers = "kafka:9092";</p><p>std::string topic = "test_topic";</p><p>std::string group_id = "cpp_consumer_group";</p><p>
</p><p>std::string errstr;</p><p>ExampleEventCb event_cb;</p><p>
</p><p>auto conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);</p><p>conf->set("bootstrap.servers", brokers, errstr);</p><p>conf->set("group.id", group_id, errstr);</p><p>conf->set("auto.offset.reset", "earliest", errstr);</p><p>conf->set("event_cb", &event_cb, errstr);</p><p>
</p><p>auto consumer = RdKafka::KafkaConsumer::create(conf, errstr);</p><p>if (!consumer) {</p><p> std::cerr << "Failed to create consumer: " << errstr << std::endl;</p><p> return 1;</p><p>}</p><p>
</p><p>consumer->subscribe({topic});</p><p>std::cout << "Consuming messages from topic " << topic << std::endl;</p><p>
</p><p>while (running) {</p><p> auto msg = consumer->consume(1000);</p><p> if (msg->err() == RdKafka::ERR_NO_ERROR) {</p><p> std::string message(reinterpret_cast<const char*>(msg->payload()), msg->len());</p><p><span style="white-space:pre"> </span>std::cout << "Received message: " << message << std::endl;</p><p> }</p><p> }</p><p> delete msg;</p><p>}</p><p>
</p><p>consumer->close();</p><p>delete consumer;</p><p>return 0;</p><p>}</p>五、Kafka 开发相关 C++ 能力
熟练使用 RAII、智能指针、异常处理
理解线程安全、异步模型（poll, callback）
能够结合 JSON/XML 配置 Kafka 客户端
编写模块化、高性能的消息收发组件
构建系统：CMake
日志：spdlog 或 glog
单元测试：gtest
JSON：nlohmann/json
————————————————
转载自：CSDN CodeWithMe
顺便给大家分享一下，民族企业大厂前后端测试捞人，待遇给的还不错，感兴趣的可以来试试！
页:
[1]