简介librdkafka 1.5.0 源码包面向 Kafka 客户端开发者与 C/C 工程师完整提供消息生产、消费、流处理及高阶消费者等核心接口同时覆盖错误处理、自动重试、分区平衡与故障恢复机制可帮助读者深入理解高性能 Kafka 客户端库的工程实现。资源共 541 个文件包含 186 个 C 源文件、95 个头文件、45 个 C 文件另有 Shell 构建脚本、Python 辅助脚本、Markdown 文档以及 cmake、conf 等构建配置文件压缩包仅 2.63MBgz 格式便于快速下载、解压与检索。包内含有核心源码与测试代码围绕 1.5.0 版本在 SASL/SCRAM 认证、TLS 加密、分区分配优化、API 扩展和日志调试等方面的改进展开从批量生产、偏移量提交到错误回调与自动重试均有涉及适合用于源码级分析、二次开发、性能调优和故障排查参考。已有 218 人学习浏览对 Kafka 客户端库感兴趣的读者可借此快速熟悉 librdkafka 的设计思路与关键机制。1. 拿到 librdkafka-1.5.0.tar.gz 之后你面对的是怎么把 Kafka 接进 C/C 世界librdkafka-1.5.0.tar.gz 是一个源码分发包解压出来的不是能直接用的二进制而是一套 Kafka 协议客户端的 C 源码需要经过 configure、make、install 三个步骤才能变成库。它解决的核心问题是你的服务是 C/C 写的不能为了一条消息去养一个 JVM但又必须和 Kafka 集群完成 produce / consume、分区、重试、心跳、版本协商这些标准动作。这个包把这些协议细节全部收进库内部业务代码只需要面对 rdkafka.h 里的十几个 API。适合的人群很明确做网关、采集探针、嵌入式进程、高性能数据管道的开发者以及负责把这些库打包进私有运行环境的基础设施工程师。换个说法如果你已经在用 Kafka 但客户端全是 Java想给 C 服务打通一条直连链路这个 tarball 就是你要的原料。它不是一个拿来即用的黑匣子而是一份需要你对自己的编译环境负责的工程起点。2. 编译前的依赖清单与 configure 参数先让 1.5.0 按你的环境生成正确的库编译前第一件事不是跑 ./configure而是把依赖列清楚。librdkafka 的 configure 是自动探测式的协议里涉及的加密、压缩、认证组件只要探测不到它不一定报错而是悄悄把对应功能禁用掉。这是新手最容易翻车的地方本地机器没装 OpenSSL 头文件configure 照样跑完make 也顺利产物也能链接可一上生产要连带 TLS 的 broker 就抓瞎。所以先把依赖过一遍再看 configure 的输出而不是只看最后一行 Done。2.1 一份最小依赖清单没装齐之前先别动 make下面是 Debian/Ubuntu 系的最小依赖参考CentOS/RHEL 系把 -dev 换成 -devel 即可。列这张表不是为了让你背包名而是为了出问题时能定位configure 探测失败的现象和真正缺的东西往往隔着半年距离。依赖探测时机缺了会怎样安装命令Debian 系gcc / make / perl必须编译直接失败build-essentialOpenSSL 头文件--enable-ssl 时日志出现 openssl not foundTLS 被禁用libssl-devzlib 头文件默认探测日志出现 zlib not found压缩协商退化zlib1g-devCyrus SASL 头文件--enable-sasl 时SASL/SCRAM 认证全部不可用libsasl2-devzstd 头文件--enable-zstd 时日志出现 zstd not foundzstd 压缩不可用libzstd-dev这里有个很多人忽略的点configure 对 ssl 和 sasl 是想用才探测。也就是说如果你根本不传 --enable-ssl它连 libssl-dev 是否存在都不关心默认产物就是一个不带 TLS 的库。这个默认行为坑过不少团队本地开发连 PLAINTEXT 的 broker 一切正常部署到要求 TLS 的集群时才重新编译来回多花半天。我一般会在 configure 这一步就把 --enable-ssl --enable-sasl 都开上无论当前是否用得到省的后面换环境重新编。lz4 是例外1.5.0 在源码里内置了 lz4 编解码默认不依赖系统库不需要额外安装。提示configure 的探测结论都打在输出的中后段养成编译完回看的习惯别只盯着最后的提示符。2.2 configure 的 6 个常用开关每个开关对应一类线上问题librdkafka-1.5.0 的 configure 支持几十个开关但日常落地用到的就这几个。下面这条命令可以看作一个生产环境的起点./configure --prefix/opt/librdkafka-1.5.0 \ --enable-ssl \ --enable-sasl \ --enable-zstd \ --enable-static \ --disable-lz4-ext逐个说--prefix 指定安装目录默认是 /usr/local但对私有部署我习惯装到独立目录避免污染系统路径后面用 pkg-config 或 rpath 指过去。--enable-ssl 开启 TLS这是线上集群的硬门槛除非你确定永远只连内网明文 broker否则必须开。--enable-sasl 开启 SASL 认证对应 Kafka 的 SASL_PLAINTEXT 和 SASL_SSL 监听方式开着的代价只是多依赖一个 libsasl2不占地。--enable-zstd 对应 broker 侧 compression.typezstd 的 topic如果你的集群启用了 zstd 压缩而客户端没开produce 时 broker 会返回不支持的压缩类型错误这条线的排查非常隐蔽。--enable-static 额外生成静态库 .a方便在交付二进制时把客户端打进同一个文件。--disable-lz4-ext 是给老平台的后悔药有些系统自带的 lz4 版本过旧或符号不干净开启后运行时可能和别的库冲突关掉让 librdkafka 用自己的内置实现最省心。如果团队统一用 CMake 管理构建1.5.0 的源码包也带 CMakeLists.txtcmake 同样能配置安装路径和 static 开关但我个人还是推荐 configure因为它在老版本 gcc、交叉编译场景下的兼容性更稳。configure 跑完会生成 Makefile 和 librdkafka.pc后面 pkg-config 能不能用就看这一步。2.3 make 与安装后的核对三行命令加两个验证make -j$(nproc) make install ldconfigmake -j 的并发数按 CPU 核数来librdkafka 编译量不算大四核机器两分钟能编完。make install 会把头文件、库、pkg-config 描述文件按 --prefix 布局放好。ldconfig 这一步要小心它只会刷新默认路径如 /lib、/usr/lib的链接器缓存如果你用 --prefix 装了独立目录ldconfig 不知道那个路径需要手动加。命令如下echo /opt/librdkafka-1.5.0/lib /etc/ld.so.conf.d/librdkafka.conf ldconfig安装完成后核对产物比看构建日志更可靠ls -l /opt/librdkafka-1.5.0/lib pkg-config --modversion librdkafka pkg-config --cflags --libs librdkafka正常输出里应该能看到 librdkafka.so.1、librdkafka.so、librdkafka.a以及 C 封装对应的 librdkafka.so.1 和 librdkafka.a。pkg-config --modversion 打印 1.5.0--cflags --libs 打印 -I 和 -L 路径及 -lrdkafka。很多编译问题到最后都归结于这两条 pkg-config 输出不对先确认它不是空再谈别的。库里同时存在 .so 和 .a链接器默认优先用 .so想强制静态见下一节。2.4 动静态库的选择与链接期定位问题动态库省内存、升级方便但部署时目标机器必须能通过 ld.so 找到 librdkafka.so.1静态库把一切都打进可执行文件适合网关、边缘节点这类不想维护依赖的环境。我一般建议开发机用动态交付物用静态。静态链接的问题是 librdkafka.a 并不是自洽的它依赖 z、ssl、crypto、sasl2 这些外部库链接时必须按依赖顺序排在你源文件的后面否则报 undefined reference。手动链接模板如下gcc -o demo demo.c -I/opt/librdkafka-1.5.0/include \ -L/opt/librdkafka-1.5.0/lib -lrdkafka \ -Wl,-rpath,/opt/librdkafka-1.5.0/lib-I 指头文件-L 指库目录-lrdkafka 的名字规则是 librdkafka.so 去掉前缀和后缀这三个一起用才能让编译器找到目标。最关键的 -Wl,-rpath 是给运行期用的程序启动时动态链接器不知道该去哪找 /opt 下的 .sorpath 相当于把路径烧进了二进制。如果不想写死路径就靠前面说的 ld.so.conf.d 方案。很多人在这里踩坑后第一反应是 export LD_LIBRARY_PATH那只对当前会话有效换个终端就失效不推荐作为交付手段。3. 用 C 接口写第一个 producer 和 consumer核心 API 与运行验证库编好了接下来是跑通第一对收发程序。librdkafka-1.5.0 的 C API 设计很直白配置对象 rd_kafka_conf_t 负责所有参数实例对象 rd_kafka_t 负责网络和服务topic 对象 rd_kafka_topic_t 负责分区路由。三步递进不多不少。下面两段代码就是完整的 producer 和 consumer可以直接抄去改地址。3.1 最小生产者从 rd_kafka_conf_new 到 rd_kafka_flush#include stdio.h #include string.h #include librdkafka/rdkafka.h int main(int argc, char **argv) { char errstr[512]; rd_kafka_conf_t *conf; rd_kafka_t *rk; rd_kafka_topic_t *rkt; const char *brokers 127.0.0.1:9092; const char *topic_name demo; /* 1. 创建配置对象设置 broker 地址 */ conf rd_kafka_conf_new(); if (rd_kafka_conf_set(conf, bootstrap.servers, brokers, errstr, sizeof(errstr)) ! RD_KAFKA_CONF_OK) { fprintf(stderr, 配置失败: %s\n, errstr); return 1; } /* 2. 创建 producer 实例成功后 conf 由库接管不能再碰 */ rk rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr)); if (!rk) { fprintf(stderr, 创建 producer 失败: %s\n, errstr); return 1; } /* 3. topic 句柄第二个参数 NULL 表示不带额外 topic 级配置 */ rkt rd_kafka_topic_new(rk, topic_name, NULL); if (!rkt) { fprintf(stderr, 创建 topic 失败\n); rd_kafka_destroy(rk); return 1; } /* 4. 发送一条消息RD_KAFKA_MSG_F_COPY 表示库内部拷贝 payload */ const char *msg hello from librdkafka-1.5.0; if (rd_kafka_produce(rkt, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY, (void *)msg, strlen(msg), NULL, 0, NULL) -1) { fprintf(stderr, produce 失败: %s\n, rd_kafka_err2str(rd_kafka_last_error())); } /* 5. 等待发送完成flush 同时驱动 delivery report 回调 */ rd_kafka_flush(rk, 10000); if (rd_kafka_outq_len(rk) 0) { fprintf(stderr, 还有 %d 条消息未确认\n, rd_kafka_outq_len(rk)); } rd_kafka_topic_destroy(rkt); rd_kafka_destroy(rk); return 0; }这段代码的逻辑链条是配置先于实例存在实例先于 topic 存在topic 先于 produce 存在。rd_kafka_conf_set 返回 RD_KAFKA_CONF_OK 才算设置有效所有拼写错误的参数名都会在这里返回错误这也是排查配置问题的最快入口。RD_KAFKA_PARTITION_UA 表示未指定分区库会根据 key 做 hash没有 key 就是轮询。RD_KAFKA_MSG_F_COPY 这个标志很关键传了它库会拷贝你的 payload 到内部 bufferproduce 返回后源内存就可以释放不传库只保留指针你必须保证源内存存活到 delivery report 触发这是 C 接口最容易出内存问题的地方。rd_kafka_flush 的第二个参数是等待毫秒数它会把内部队列里所有未确认消息推到 broker 并等待回执超时返回时 outq_len 大于 0 说明有消息没送出去。3.2 最小消费者subscribe 与 poll 的驱动关系#include stdio.h #include string.h #include librdkafka/rdkafka.h int main(int argc, char **argv) { char errstr[512]; rd_kafka_conf_t *conf; rd_kafka_t *rk; rd_kafka_topic_partition_list_t *topics; rd_kafka_message_t *msg; const char *brokers 127.0.0.1:9092; const char *group demo-group; const char *topic_name demo; conf rd_kafka_conf_new(); rd_kafka_conf_set(conf, bootstrap.servers, brokers, errstr, sizeof(errstr)); /* 消费者必须设 group.id这是加入消费组的身份 */ rd_kafka_conf_set(conf, group.id, group, errstr, sizeof(errstr)); /* earliest 表示从头消费只对新 group 或重置过的 group 生效 */ rd_kafka_conf_set(conf, auto.offset.reset, earliest, errstr, sizeof(errstr)); rk rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr)); if (!rk) { fprintf(stderr, consumer 创建失败: %s\n, errstr); return 1; } /* 构建订阅列表PARTITION_UA 表示订阅整个 topic */ topics rd_kafka_topic_partition_list_new(1); rd_kafka_topic_partition_list_add(topics, topic_name, RD_KAFKA_PARTITION_UA); if (rd_kafka_subscribe(rk, topics) ! RD_KAFKA_RESP_ERR_NO_ERROR) { fprintf(stderr, subscribe 失败: %s\n, rd_kafka_err2str(rd_kafka_last_error())); return 1; } rd_kafka_topic_partition_list_destroy(topics); /* poll 是消费的总驱动拉消息、跑心跳、触发 rebalance 都靠它 */ for (int i 0; i 10; i) { msg rd_kafka_consumer_poll(rk, 1000); if (!msg) continue; /* 超时无消息不是错误 */ if (msg-err) { printf(消息错误: %s\n, rd_kafka_err2str(msg-err)); } else { printf(收到 [%s-%d] offset%ld: %.*s\n, msg-topic_name, msg-partition, (long)msg-offset, (int)msg-len, (char *)msg-payload); } rd_kafka_message_destroy(msg); /* 不销毁就是内存泄漏 */ } rd_kafka_consumer_close(rk); rd_kafka_destroy(rk); return 0; }consumer 和 producer 最大的认知差异在 poll 的语义producer 的 poll 只是取 delivery reportconsumer 的 poll 身兼多职包括从 broker 拉批次、给协调器发心跳、处理 rebalance 回调。poll 长时间不调用group 协调器会判定会话超时直接把你的消费者踢出分组。auto.offset.reset 的语义坑写在注释里了它只对新 group 或没有提交过 offset 的分区生效一个已经跑过的 group ID 不会因为改成 earliest 就从头读这是后面避坑章的重点。收到消息后必须 rd_kafka_message_destroy这条不执行每条消息泄漏一份 payload跑一天的常驻进程内存爬升非常可观。3.3 编译命令与三种链接姿势gcc -o demo_producer demo_producer.c $(pkg-config --cflags --libs librdkafka) gcc -o demo_consumer demo_consumer.c $(pkg-config --cflags --libs librdkafka)这是最省心的姿势pkg-config 会把 -I、-L、-lrdkafka 一次性展开。注意 $( ) 的位置命令替换必须出现在源文件之后如果写成gcc $(pkg-config ...) demo.c-lrdkafka 会在源文件之前出现静态链接时就会复现 undefined reference。C 封装的链接是另一条线g -stdc11 -o demo demo.cpp $(pkg-config --cflags --libs librdkafka) \ -lrdkafka3.4 用 tarball 自带的 examples 做冒烟验证源码包解压后自带 examples 目录里面是官方维护的 producer、consumer 示例这是验证库是否可用的最快途径。常见做法是进 examples 目录直接 make然后先起消费者再起生产者观察消息能不能打出来。用法上只需要把示例里的 brokers 和 topic 改成你自己的。如果示例能跑通说明 configure 开关、链接姿势、broker 连通性这三关都过了如果再往后遇到业务代码的问题可以排除库本身。4. 关键配置参数与 batch、超时、commit 的行为边界librdkafka 的参数上百个但真正需要手调的没几个。这一章把 producer、consumer 两侧最影响行为边界的参数讲透参数背后是库的调度机理不理解机理就只会照着网上抄值换一个流量模型就失效。4.1 producer 侧acks、linger.ms、message.timeout.ms、queue.buffering.max.messages参数默认值作用边界调优建议acks-10 不等确认1 等 leader 确认-1/all 等 ISR 全部确认数据敏感用 alllinger.ms5攒批等待时间不影响单条延迟上限高吞吐可调到 10~20message.timeout.ms300000一条消息从入队到 broker 确认的最大等待按业务容忍度调别用默认 5 分钟queue.buffering.max.messages100000本地队列上限满了 produce 返回 -1按内存和消费速度权衡参数设置的代码块如下rd_kafka_conf_set(conf, acks, all, errstr, sizeof(errstr)); rd_kafka_conf_set(conf, linger.ms, 10, errstr, sizeof(errstr)); rd_kafka_conf_set(conf, message.timeout.ms, 30000, errstr, sizeof(errstr)); rd_kafka_conf_set(conf, queue.buffering.max.messages, 100000, errstr, sizeof(errstr));linger.ms 的语义是为了攒批最多等多久不是每条消息延迟多久。内部线程会边攒边发只要 batch 攒够或者达到 linger 时间就会发出。message.timeout.ms 是队列里一条消息的寿命上限超过后库会丢弃它并触发 error 回调所以它不是重试超时而是放弃时间。想精细控制重试次数还要关注 retries 参数不设置时默认会一直重试到 message.timeout.ms 耗尽。queue.buffering.max.messages 是背压阀门消费方跟不上时produce 会返回 -1rd_kafka_last_error 报 RD_KAFKA_RESP_ERR__QUEUE_FULL业务方该把这次调用当成失败处理而不是忽略。4.2 consumer 侧group.id、auto.offset.reset、enable.auto.commit、session.timeout.ms参数默认值作用边界建议group.id无消费组身份决定分区分配和 offset 存储按业务语义命名auto.offset.resetlargest新 group / 无提交 offset 时的起点测试用 earliest线上按产品enable.auto.committrue是否自动提交已 poll 的 offset精确一次场景关掉session.timeout.ms10000不心跳的最大间隔超时被踢出组并 rebalance别小于 6000会给 GC 留空间rd_kafka_conf_set(conf, group.id, group, errstr, sizeof(errstr)); rd_kafka_conf_set(conf, enable.auto.commit, true, errstr, sizeof(errstr)); rd_kafka_conf_set(conf, auto.commit.interval.ms, 5000, errstr, sizeof(errstr)); rd_kafka_conf_set(conf, session.timeout.ms, 10000, errstr, sizeof(errstr)); rd_kafka_conf_set(conf, heartbeat.interval.ms, 3000, errstr, sizeof(errstr));enable.auto.commit 的坑在于它的提交时机库每 auto.commit.interval.ms 会提交当前已通过 poll 返回的最高 offset而不是业务已处理完的 offset。如果你的业务在 poll 之后做了耗时操作比如写数据库操作还没完成自动提交就到了进程 crash 后会从已提交的 offset 继续读那些没写库的消息就丢了。要收紧语义把 enable.auto.commit 设为 false在处理成功后手动调 rd_kafka_commit_message。session.timeout.ms 和 heartbeat.interval.ms 的比例关系也值得注意心跳间隔一般是会话超时的三分之一留足网络抖动余量调太高会让 group 的 rebalance 变迟钝。4.3 排查用参数debug、log_level、statistics.interval.ms这三兄弟是定位问题的三件套平时不开出问题第一时间打开。rd_kafka_conf_set(conf, debug, broker,topic,msg, errstr, sizeof(errstr)); rd_kafka_conf_set(conf, log_level, 7, errstr, sizeof(errstr)); rd_kafka_conf_set(conf, statistics.interval.ms, 5000, errstr, sizeof(errstr));debug 参数接受逗号分隔的模块名broker 打印与 broker 的协议交互、topic 打印元数据变更、msg 打印消息收发路径三者叠加能看清一条消息从入队到确认的完整生命周期。log_level 7 对应 DEBUG 级配合顶层 log_cb 回调可以接入自有日志体系否则默认打到 stderr。statistics.interval.ms 设置后会周期性触发 stats_cb 回调并传入 JSON 统计快照这个功能是生产体检的核心第 6 章会展开。这三个参数的公共原则是排查完立刻关掉特别是 debug开着会有可感知的吞吐下降。4.4 线程模型谁在干活poll 为什么不能省librdkafka 1.5.0 内部有独立的后台线程池负责 socket 读写、协议解析、分区元数据同步这部分不需要业务操心。但用户态的两个队列必须由业务线程驱动producer 的 delivery report 队列和 consumer 的消费队列。delivery report 是异步确认信号你不调 rd_kafka_poll报告就积压在内部时间长了队列溢出消息虽然送出去了但你的代码不知道成功还是失败。consumer 更严重不调 consumer_poll心跳没人发coordinator 把你踢出组rebalance 会重新分配分区。很多消费者跑一会儿就停的线上事故根因就是 poll 循环里插了阻塞操作比如同步 HTTP 调用。正确姿势是 poll 永远在独立线程业务处理放别处处理完再回来 poll。5. 避坑手册编译与运行的 5 个高频踩坑记录这一章是血泪经验的压缩包。每一条都按现象、原因、解决三段写遇到直接对照能省半天排查时间。5.1 configure 通过、运行却报告 SSL 不可用的静默失败现象configure 全程绿灯make 顺利程序能起但连开启 TLS 的 broker 时日志报 SSL 相关错误或 rd_kafka_conf_set 设置 security.protocolSASL_SSL 直接返回配置失败。原因configure 只有在显式传 --enable-ssl 时才探测 OpenSSL你没传它就不报错、也不编入产物里根本没有 TLS 支持。解决回 configure 那一步确认命令里带 --enable-ssl 和 --enable-sasl重编重装。用 pkg-config 验证不了这个因为 pkg-config 不管功能是否编入最简单的确认方式是编译时用gcc ... -lrdkafka -lssl如果链接器不报错才说明 ssl 功能真的在里面。5.2 链接顺序不对导致的 undefined reference现象编译命令里明明写了 -lrdkafka却报 undefined reference to rd_kafka_new。原因gcc 对静态库的解析是顺序相关的-lrdkafka 出现在源文件之前时链接器还没遇到未定义符号等遇到时已经过去不回头了。解决把库参数放到源文件之后静态链接时还要把 z、ssl、crypto、sasl2 按依赖顺序放在 -lrdkafka 之后。简单排查法用 pkg-config --libs librdkafka 展开确认输出在编译命令末尾。5.3 换机器后 runtime 找不到 librdkafka.so.1现象程序在编译机跑得好好的拷到别的机器执行报 error while loading shared libraries: librdkafka.so.1: cannot open shared object file。原因你编的包是动态链接目标机的 ld.so 不知道去哪找这个 so。解决三选一目标机写 ld.so.conf.d 并 ldconfig链接时加 -Wl,-rpath 把路径烧进二进制或者直接静态链接交付。第三种最省事尤其适合容器镜像省一层运行时依赖管理。5.4 consumer 空转不消费group ID 的历史包袱现象poll 一直返回 NULL程序日志干净没有任何报错手动到命令行客户端却能读到同一 topic 的数据。原因最常见三个一是新 group 配了 latest 且 producer 在 consumer 启动前就发完了数据二是 group ID 在别的环境跑过offset 提交点已经跳到最新位置改 auto.offset.reset 不生效三是组里其他消费者已把全部分区占走。解决先启动消费者再启动生产者做连通性测试测试就用全新的 group ID 加 earliest排查时打印实际分配为准。打印分配rd_kafka_topic_partition_list_t *assigned rd_kafka_assignment(rk); for (int i 0; i assigned-cnt; i) { printf(分配: %s-%d\n, assigned-elems[i].topic, assigned-elems[i].partition); } rd_kafka_topic_partition_list_destroy(assigned);5.5 broker 版本协商失败api.version.request 的双刃剑现象produce 返回成功但消息没进 topic日志里出现 unsupported version 或 broker 行为异常。原因librdkafka 1.5.0 默认向 broker 发 ApiVersions 请求来协商协议版本老版本的 broker 不认识这个请求协商就直接失败客户端退不回来。解决给配置加上api.version.requestfalse和broker.version.fallback0.10.0让客户端按固定版本协商不依赖探测。对现在的主流集群这不是问题但如果你维护的是长期没升级的老集群这组参数要提前写进配置模板。这个坑的排查难点在于它不报硬错误produce 的返回码是正常的只有看服务端 log 或开启 debugbroker 才能看到协商痕迹。6. 用统计回调和 debug 日志给客户端做运行体检库编好了、收发通了最后一步是把运行状态可视化。librdkafka 1.5.0 自带的统计回调是一个被低估的利器只要设了 statistics.interval.ms库就会按周期生成一份 JSON 统计快照里面几乎包含你排查生产问题需要的所有指标。注册方式很简单static int stats_cb(rd_kafka_t *rk, char *json, size_t json_len, void *opaque) { /* 实际项目里这里应把 json 写到监控 agent 的输入管道 */ fwrite(json, 1, json_len, stderr); fputc(\n, stderr); return 0; } rd_kafka_conf_set_stats_cb(conf, stats_cb); rd_kafka_conf_set(conf, statistics.interval.ms, 10000, errstr, sizeof(errstr));配合 debugbroker 一起用能看到两个维度的信息debug 日志是实时的协议交互流水统计 JSON 是周期性的健康快照。读快照时我最关注三个字段顶层 msg_cnt 和 msg_bytes 反映本地队列积压这两个值持续上涨说明消费端处理不过来或 broker 侧出了瓶颈brokers 数组里的 rtt 反映网络往返latency 异常说明链路不稳。具体做法是起消费者跑 10 分钟生产者灌数据然后观察统计 JSON 里 msg_cnt 是否稳定在低位、produce 侧 outq_len 是否归零两个都正常说明整条链路是健康的。如果 JSON 里某个字段看不懂不用死记1.5.0 的统计输出字段名和配置项是对应的按名字查配置即可。我个人的习惯是每次切换 librdkafka 版本都先编译一个十行代码的自检程序跑通收发再谈接入业务这套体检流程会帮你把库的问题和业务的问题彻底分开。希望帮到你。本文还有配套的精品资源点击获取