librdkafka 开源 Kafka 客户端库深度解析

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

1. 背景

1.1 为什么需要 Kafka 客户端库

Apache Kafka 是分布式流处理平台的核心组件，以「分布式提交日志（distributed commit log）」模型著称：消息按分区（Partition）顺序追加、多副本冗余、消费者按 offset 游标自主拉取。它在日志采集、指标上报、事件驱动架构、数据管道中几乎是事实标准。

Kafka 官方提供了 Java 客户端，但工业数采、嵌入式边缘网关、高性能 C++ 服务等场景无法引入 JVM。librdkafka 正是为此而生的：

• 纯 C/C++ 实现，零 JVM 依赖，可在 ARM、x86 嵌入式 Linux、Windows 上直接编译；
• 由 Confluent 维护，是 Kafka 官方钦定的 C/C++ 客户端（Java 之外社区使用最广）；
• 吞吐极高（百万级 msg/s 量级）、延迟低，支持 Kafka 0.8 ~ 3.x 全版本线；
• 同时提供 C API（rdkafka.h）与 C++ 封装（rdkafkacpp.h），本文以 C++ API 为主线。

1.2 librdkafka 的定位

2. API 说明

2.1 安装与 CMake 集成

# Ubuntu
sudo apt install librdkafka-dev
# Windows (vcpkg)
vcpkg install librdkafka

find_package(RdKafka REQUIRED)
target_link_libraries(your_target PRIVATE RdKafka::rdkafka++)

注意：C++ API 的 CMake target 是 RdKafka::rdkafka++（仅链接 librdkafka++），若只用 C API 则链接 RdKafka::rdkafka。

2.2 核心类与头文件

头文件 <librdkafka/rdkafkacpp.h> 提供了以下核心类型：

2.3 关键配置项（Conf::set 的 key）

通用

生产者

消费者

2.4 回调机制

librdkafka 是回调驱动模型，核心回调：

• DeliveryReportCb::dr_cb：每条消息 produce 后由库内部线程触发，携带 Message，通过 msg->err() 判断投递成功（ERR_NO_ERROR）或失败（ERR_MSG_TIMED_OUT 等）。
• EventCb::event_cb：错误、日志、统计事件统一入口，RdKafka::Event 提供 type()、severity()、err()、str()。
• RebalanceCb::rebalance_cb：消费组发生 rebalance（加入/离开/分区分配变更）时触发，需要配合 RdKafka::ERR__ASSIGN_PARTITIONS / ERR__REVOKE_PARTITIONS 调用 assign() / unassign()。

所有回调都在调用 poll() / flush() / consume() 的线程（即你主动驱动）中执行。必须定期调用这些驱动函数，否则回调永远不触发——这是新手最常见的误区。

3. 详细使用说明

3.1 生产者：发送一条消息（含投递报告）

#include <librdkafka/rdkafkacpp.h>
#include <iostream>
#include <string>

// 投递报告回调：每条消息的发送结果都会回到这里
class DrCb : public RdKafka::DeliveryReportCb {
public:
void dr_cb(RdKafka::Message &msg) override {
if (msg.err() == RdKafka::ERR_NO_ERROR) {
std::cout << "[OK] " << msg.topic_name() << "[" << msg.partition()
<< "] @ " << msg.offset() << std::endl;
} else {
std::cerr << "[FAIL] " << msg.errstr() << std::endl;
}
}
};

int main() {
// 1. 创建配置
RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
DrCb dr_cb;
if (conf->set("bootstrap.servers", "192.168.1.10:9092", errstr) != RdKafka::Conf::CONF_OK ||
conf->set("acks", "all", errstr) != RdKafka::Conf::CONF_OK ||
conf->set("dr_cb", &dr_cb, errstr) != RdKafka::Conf::CONF_OK) {
std::cerr << "conf error: " << errstr << std::endl;
return 1;
}

// 2. 创建生产者
RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
delete conf; // create 之后 Conf 可释放
if (!producer) {
std::cerr << "producer create failed: " << errstr << std::endl;
return 1;
}

// 3. 发送消息（payload/key 为 void* + size，会被库拷贝）
std::string payload = R"({"device":"CNC-01","value":42.5})";
std::string key = "CNC-01";
RdKafka::ErrorCode err = producer->produce(
"factory.telemetry", // topic
RdKafka::Topic::PARTITION_UA, // 未指定分区，按 key 哈希
RdKafka::Producer::RK_MSG_COPY, // 拷贝 payload
const_cast<char*>(payload.data()), payload.size(),
const_cast<char*>(key.data()), key.size(),
0, nullptr);

if (err != RdKafka::ERR_NO_ERROR) {
std::cerr << "produce failed: " << RdKafka::err2str(err) << std::endl;
delete producer;
return 1;
}

// 4. 驱动回调 + 等待队列清空（必须！）
producer->flush(10000); // 阻塞至多 10s，期间驱动投递报告回调
std::cout << "outq_len after flush: " << producer->outq_len() << std::endl;

delete producer;
return 0;
}

关键点：produce() 只是把消息放进内部发送队列，真正的网络发送与投递报告都由 flush()/poll() 驱动。

3.2 生产者：按 key 路由到指定分区

// 指定分区发送
err = producer->produce("factory.telemetry", 0 /* partition */,
RdKafka::Producer::RK_MSG_COPY,
const_cast<char*>(payload.data()), payload.size(),
nullptr, 0, 0, nullptr);

不指定分区（PARTITION_UA）时，librdkafka 默认按 key 做 murmur2 哈希路由（相同 key 进同一分区，保证顺序）；无 key 则轮询。

3.3 消费者：消费组模式（手动提交）

#include <librdkafka/rdkafkacpp.h>

int main() {
RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
std::string errstr;
conf->set("bootstrap.servers", "192.168.1.10:9092", errstr);
conf->set("group.id", "factory-collector", errstr);
conf->set("enable.auto.commit", "false", errstr); // 手动提交
conf->set("auto.offset.reset", "earliest", errstr);
conf->set("enable.partition.eof", "false", errstr);

RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);
delete conf;
if (!consumer) { std::cerr << errstr << std::endl; return 1; }

// 订阅主题
std::vector<std::string> topics = {"factory.telemetry"};
RdKafka::ErrorCode err = consumer->subscribe(topics);
if (err != RdKafka::ERR_NO_ERROR) {
std::cerr << "subscribe failed: " << RdKafka::err2str(err) << std::endl;
return 1;
}

int msg_cnt = 0;
while (msg_cnt < 100) { // 示例：消费 100 条退出
RdKafka::Message *msg = consumer->consume(1000 /* 阻塞最多 1s */);
if (!msg) continue;

switch (msg->err()) {
case RdKafka::ERR_NO_ERROR: {
std::string payload(static_cast<const char*>(msg->payload()), msg->len());
std::cout << "got: " << payload
<< " [p" << msg->partition() << "@" << msg->offset() << "]" << std::endl;
++msg_cnt;
// 处理完成后提交 offset（异步提交，下一轮再校验）
consumer->commitAsync(msg);
delete msg;
break;
}
case RdKafka::ERR__TIMED_OUT:
delete msg;
break;
default:
std::cerr << "consume error: " << msg->errstr() << std::endl;
delete msg;
break;
}
}

consumer->close(); // 关闭前会尝试提交未提交的 offset
delete consumer;
return 0;
}

3.4 消费者：seek 回退到指定 offset 重放

// 场景：业务发现第 5000 条处理出错，需要从 4800 重新消费
RdKafka::TopicPartition *tp = RdKafka::TopicPartition::create(
"factory.telemetry", 0, 4800);
RdKafka::TopicPartitionList tpl;
tpl.push_back(tp);
consumer->seek(tpl, 0);
delete tp;

3.5 消费者：手动 assign 指定分区（绕过消费组）

RdKafka::TopicPartition *tp = RdKafka::TopicPartition::create("factory.telemetry", 0, RdKafka::Topic::OFFSET_STORED);
RdKafka::TopicPartitionList tpl;
tpl.push_back(tp);
consumer->assign(tpl); // 不再使用 subscribe，自己接管分区
delete tp;

适用场景：单消费者固定消费某几个分区（如每个数采网关只消费自己设备的分区），不参与组再平衡。

3.6 带 SASL 认证的连接

conf->set("security.protocol", "sasl_plaintext", errstr);
conf->set("sasl.mechanisms", "SCRAM-SHA-512", errstr);
conf->set("sasl.username", "collector", errstr);
conf->set("sasl.password", "your-password", errstr);

3.7 再平衡回调（RebalanceCb）完整示例

class RebalanceCb : public RdKafka::RebalanceCb {
public:
void rebalance_cb(RdKafka::KafkaConsumer *consumer,
RdKafka::ErrorCode err,
std::vector<RdKafka::TopicPartition*> &partitions) override {
if (err == RdKafka::ERR__ASSIGN_PARTITIONS) {
consumer->assign(partitions); // 拿到分区
} else if (err == RdKafka::ERR__REVOKE_PARTITIONS) {
consumer->unassign(); // 分区被回收，先落盘再放弃
} else {
std::cerr << "rebalance error: " << RdKafka::err2str(err) << std::endl;
}
}
};
// 注册：conf->set("rebalance_cb", &rebalance_cb, errstr);

4. 常错点 / 坑

4.1 生产端高频坑

4.2 消费端高频坑

4.3 连接与配置坑

4.4 事务与高级场景坑

5. 总结

5.1 适用场景

5.2 生态衔接建议（与既有篇目联动）

• 采集端：libmodbus（Modbus 现场总线）→ 数采程序（C++）→ librdkafka（消息总线）；
• 传输底座：Boost.Interprocess（同机进程间共享内存）可作为采集进程与上报进程之间的本地缓冲；
• 跨语言消费：pybind11 可将 librdkafka 封装为 Python 模块；Go 侧可用 confluent-kafka-go（底层同样是 librdkafka）接入同一 Kafka 集群。

5.3 选型决策树

需要消息中间件？
├─ 单机/轻量/无需持久化 → ZeroMQ
├─ 同机进程间共享数据 → Boost.Interprocess（共享内存）
├─ 需要持久化 + 消费组 + 跨主机分布式
│ └─ Kafka（librdkafka）
└─ 需要协议网关而非消息系统 → libmodbus / gRPC / MQTT

5.4 学习路径建议

6. FAQ 速查表