Apache Kafka 是一款高吞吐、低延迟、可持久化、分区有序的分布式流式消息队列相比于 RocketMQKafka 在海量日志采集、实时数据流、大数据流式计算、海量埋点上报场景优势极大。Java 生态 Kafka 教程泛滥但高性能 C 网关、边缘采集服务、日志转发服务、游戏后端、大数据采集客户端均需要使用官方librdkafka进行开发。librdkafka 是 Kafka 官方推荐的 C/C SDK具备极致吞吐、内存占用低、异步模型成熟、集群高可用等特性是工业级 C 接入 Kafka 的唯一首选方案。本文从零带你落地 Kafka C 全套开发能力包含环境编译、基础生产消费、同步/异步发送、批量发送、分区有序、消息重试、失败回调、消费者位移管理、集群高可用、全局单例封装、生产场景选型与避坑所有代码可直接编译上线。一、Kafka C SDK 核心概述1.1 什么是 librdkafkalibrdkafka是 Apache Kafka 官方开源的高性能 C/C 客户端库底层基于异步事件驱动模型支持 Kafka 全部核心协议跨平台、无虚拟机依赖是目前性能最强、稳定性最高的 Kafka 客户端。1.2 核心支持能力✅ 同步发送、异步发送、批量聚合发送✅ 分区策略、自定义分区投递、局部顺序消息✅ 消费者自动分区负载均衡、重平衡机制✅ 手动/自动 offset 位移提交✅ 消息发送失败重试、异常回调捕获✅ Kafka 集群多节点高可用自动容错✅ 批量消费、消息头属性透传✅ 全局单例客户端封装、连接池复用思想1.3 Kafka vs RocketMQ 适用场景差异Kafka极致高吞吐、海量日志、数据流、埋点、大数据实时计算侧重流式处理RocketMQ业务消息、订单、事务、死信、重试队列侧重业务可靠性1.4 C SDK 核心优势无 JVM 依赖无 GC 抖动延迟极其稳定事件驱动异步模型单机支撑百万级 QPS内存占用极低适合长期驻留后台服务完美适配 C 高性能服务技术栈二、环境搭建 工程配置2.1 系统依赖安装sudo apt update sudo apt install git cmake gcc g make libssl-dev zlib1g-dev -y2.2 编译安装 librdkafka# 拉取源码 git clone https://github.com/edenhill/librdkafka.git cd librdkafka mkdir build cd build # 编译安装 cmake -DCMAKE_BUILD_TYPERelease .. make -j$(nproc) sudo make install sudo ldconfig2.3 项目目录结构kafka_cpp_demo/ ├── CMakeLists.txt └── main.cpp2.4 完整 CMakeLists.txt 配置cmake_minimum_required(VERSION 3.16) project(kafka_cpp_demo) set(CMAKE_CXX_STANDARD 17) set(CMAKE_CXX_STANDARD_REQUIRED ON) # 头文件与库路径 include_directories(/usr/local/include) link_directories(/usr/local/lib) add_executable(kafka_demo main.cpp) # 链接kafka依赖 target_link_libraries(kafka_demo rdkafka rdkafka pthread ssl z )三、Kafka 基础核心概念BrokerKafka 服务节点集群多节点部署Topic消息主题业务数据隔离单元Partition分区Kafka 并发与有序性核心单元Offset消息位移消费者消费标记ConsumerGroup消费者组实现负载均衡四、基础生产者实战同步/异步/批量头文件统一引入下文所有代码共用#include iostream #include string #include vector #include chrono #include thread #include librdkafka/rdkafkacpp.h using namespace std;4.1 同步消息发送业务可靠消息同步发送阻塞等待 broker 确认适合业务可靠、不允许丢失的消息场景。void KafkaSyncProduce() { // 配置项 string errstr; RdKafka::Conf* conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); // Kafka集群地址单机填写单个集群用逗号分隔 conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); // 创建生产者 RdKafka::Producer* producer RdKafka::Producer::create(conf, errstr); if (!producer) { cerr Producer 创建失败 errstr endl; return; } string topic cpp_kafka_test_topic; string msg Kafka C 同步消息内容; // 同步发送 RdKafka::ErrorCode resp producer-produce( topic, RdKafka::PARTITION_UA, RdKafka::MSG_COPY, (void*)msg.c_str(), msg.size(), nullptr, 0, nullptr ); if (resp ! RdKafka::ERR_NO_ERROR) { cerr 消息发送失败 RdKafka::err2str(resp) endl; } else { cout 同步消息发送成功 endl; } // 刷新缓冲区确保消息发出 producer-flush(1000); delete producer; delete conf; }4.2 异步消息发送高吞吐日志/埋点异步发送不阻塞业务线程通过回调返回发送结果适合海量埋点、日志上报、监控数据等高吞吐场景。// 异步发送回调 class DeliveryReportCb : public RdKafka::DeliveryReportCb { public: void dr_cb(RdKafka::Message msg) override { if (msg.err()) { cerr 异步消息发送失败 msg.errstr() endl; } else { cout 异步消息发送成功, partition: msg.partition() , offset: msg.offset() endl; } } }; void KafkaAsyncProduce() { string errstr; RdKafka::Conf* conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); // 开启异步队列超时刷新 conf-set(queue.buffering.max.ms, 5, errstr); static DeliveryReportCb dr_cb; conf-set(dr_cb, dr_cb, errstr); RdKafka::Producer* producer RdKafka::Producer::create(conf, errstr); if (!producer) { cerr Producer 创建失败 errstr endl; return; } string topic cpp_kafka_test_topic; string msg Kafka C 异步高吞吐消息; // 异步投递 producer-produce( topic, RdKafka::PARTITION_UA, RdKafka::MSG_COPY, (void*)msg.c_str(), msg.size(), nullptr, 0, nullptr ); // 轮询触发回调 producer-poll(0); this_thread::sleep_for(chrono::milliseconds(20)); delete producer; delete conf; }4.3 批量消息发送极致吞吐优化批量发送合并多条消息一次网络IO大幅提升吞吐是 Kafka 大数据采集场景标准用法。void KafkaBatchProduce() { string errstr; RdKafka::Conf* conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); static DeliveryReportCb dr_cb; conf-set(dr_cb, dr_cb, errstr); // 批量聚合参数 conf-set(batch.size, 16384, errstr); conf-set(linger.ms, 5, errstr); RdKafka::Producer* producer RdKafka::Producer::create(conf, errstr); string topic cpp_kafka_test_topic; for (int i 0; i 20; i) { string msg 批量消息数据_ to_string(i); producer-produce( topic, RdKafka::PARTITION_UA, RdKafka::MSG_COPY, (void*)msg.c_str(), msg.size(), nullptr, 0, nullptr ); producer-poll(0); } producer-flush(1000); delete producer; delete conf; cout 批量消息发送完成 endl; }4.4 分区有序消息发送业务顺序保障Kafka同一分区内消息有序通过自定义 key 哈希路由到固定分区实现订单、流程类有序消息。void KafkaOrderProduce() { string errstr; RdKafka::Conf* conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); static DeliveryReportCb dr_cb; conf-set(dr_cb, dr_cb, errstr); RdKafka::Producer* producer RdKafka::Producer::create(conf, errstr); string topic cpp_kafka_order_topic; string order_key ORDER_10001; for (int i 1; i 5; i) { string msg 订单流程步骤_ to_string(i); producer-produce( topic, RdKafka::PARTITION_UA, RdKafka::MSG_COPY, (void*)msg.c_str(), msg.size(), order_key.c_str(), order_key.size(), nullptr ); producer-poll(0); } producer-flush(1000); delete producer; delete conf; cout 顺序消息发送完成 endl; }五、基础消费者实战自动位移提交void KafkaConsume() { string errstr; RdKafka::Conf* conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); RdKafka::Conf* tconf RdKafka::Conf::create(RdKafka::CONF_TOPIC); // 集群地址 conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); // 消费者组 conf-set(group.id, cpp_kafka_consumer_group, errstr); // 自动提交offset conf-set(enable.auto.commit, true, errstr); conf-set(auto.commit.interval.ms, 1000, errstr); // 首次消费从头开始 conf-set(auto.offset.reset, earliest, errstr); RdKafka::KafkaConsumer* consumer RdKafka::KafkaConsumer::create(conf, tconf, errstr); if (!consumer) { cerr 消费者创建失败 errstr endl; return; } // 订阅topic vectorstring topics {cpp_kafka_test_topic}; consumer-subscribe(topics); cout Kafka消费者启动成功等待消息... endl; while (true) { RdKafka::Message* msg consumer-consume(1000); if (msg-err() RdKafka::ERR_NO_ERROR) { cout 收到消息 (char*)msg-payload() partition: msg-partition() offset: msg-offset() endl; } delete msg; } consumer-close(); delete consumer; delete conf; delete tconf; }六、企业级高阶扩展功能生产必备6.1 Kafka 集群高可用配置多节点自动容错生产环境必须配置多 broker 节点SDK 自动负载均衡、故障节点剔除、自动重连彻底消除单点故障。void KafkaClusterProduce() { string errstr; RdKafka::Conf* conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); // 多节点集群地址逗号分隔 string cluster_addr 192.168.1.100:9092,192.168.1.101:9092,192.168.1.102:9092; conf-set(bootstrap.servers, cluster_addr, errstr); static DeliveryReportCb dr_cb; conf-set(dr_cb, dr_cb, errstr); RdKafka::Producer* producer RdKafka::Producer::create(conf, errstr); string msg Kafka集群高可用测试消息; producer-produce(cpp_kafka_cluster_topic, RdKafka::PARTITION_UA, RdKafka::MSG_COPY, (void*)msg.c_str(), msg.size(), nullptr, 0, nullptr); producer-flush(1000); delete producer; delete conf; cout 集群消息发送成功 endl; }6.2 手动 Offset 提交精准消费保障自动提交存在消息丢失/重复消费风险核心业务建议业务处理成功后手动提交位移。void KafkaManualOffsetConsume() { string errstr; RdKafka::Conf* conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); RdKafka::Conf* tconf RdKafka::Conf::create(RdKafka::CONF_TOPIC); conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); conf-set(group.id, cpp_kafka_manual_group, errstr); // 关闭自动提交 conf-set(enable.auto.commit, false, errstr); conf-set(auto.offset.reset, earliest, errstr); RdKafka::KafkaConsumer* consumer RdKafka::KafkaConsumer::create(conf, tconf, errstr); consumer-subscribe({cpp_kafka_test_topic}); cout 手动位移消费者启动成功 endl; while (true) { RdKafka::Message* msg consumer-consume(1000); if (msg-err() RdKafka::ERR_NO_ERROR) { // 模拟业务处理 cout 业务处理消息 (char*)msg-payload() endl; // 业务成功手动提交offset consumer-commitSync(*msg); } delete msg; } consumer-close(); delete consumer; delete conf; delete tconf; }6.3 C 全局单例 Kafka 生产者封装生产最优方案Kafka Producer 是重量级对象禁止频繁创建销毁。封装线程安全懒汉单例全局复用连接避免句柄泄漏、连接风暴。#include mutex #include memory class KafkaGlobalProducer { public: static KafkaGlobalProducer Instance() { static KafkaGlobalProducer ins; return ins; } // 初始化全局生产者 bool Init(const string broker_list) { lock_guardmutex lock(mtx); if (m_producer) return true; string errstr; m_conf.reset(RdKafka::Conf::create(RdKafka::CONF_GLOBAL)); m_conf-set(bootstrap.servers, broker_list, errstr); m_conf-set(queue.buffering.max.ms, 5, errstr); m_conf-set(batch.size, 16384, errstr); static DeliveryReportCb dr_cb; m_conf-set(dr_cb, dr_cb, errstr); m_producer.reset(RdKafka::Producer::create(m_conf.get(), errstr)); if (!m_producer) { cerr 全局生产者初始化失败 errstr endl; return false; } cout Kafka全局单例生产者初始化成功 endl; return true; } // 通用发送接口 bool SendMsg(const string topic, const string body, const string key ) { if (!m_producer) return false; RdKafka::ErrorCode ret m_producer-produce( topic, RdKafka::PARTITION_UA, RdKafka::MSG_COPY, (void*)body.c_str(), body.size(), key.empty() ? nullptr : key.c_str(), key.size(), nullptr ); m_producer-poll(0); return ret RdKafka::ERR_NO_ERROR; } // 退出释放资源 void Shutdown() { lock_guardmutex lock(mtx); if (m_producer) { m_producer-flush(1000); m_producer.reset(); } if (m_conf) m_conf.reset(); } private: KafkaGlobalProducer() default; ~KafkaGlobalProducer() { Shutdown(); } KafkaGlobalProducer(const KafkaGlobalProducer) delete; KafkaGlobalProducer operator(const KafkaGlobalProducer) delete; mutex mtx; unique_ptrRdKafka::Conf m_conf; unique_ptrRdKafka::Producer m_producer; }; // 单例调用示例 void KafkaSingletonTest() { // 服务启动初始化一次 KafkaGlobalProducer::Instance().Init(127.0.0.1:9092); // 业务任意位置发送消息 KafkaGlobalProducer::Instance().SendMsg(cpp_kafka_global_topic, 全局单例Kafka消息); }6.4 消息重试 死信队列(DLQ) 生产级完整实现Kafka 原生无内置重试队列与死信队列机制生产环境需手动实现重试策略死信投递解决业务瞬时失败重试、异常消息隔离、数据兜底修复问题。本节实现工业级方案消费失败阶梯重试、最大重试次数拦截、异常消息转入死信Topic、死信独立消费、人工修复重发完全对齐企业生产规范。6.4.1 核心设计思路自定义消息重试次数标记限定最大重试次数默认8次业务异常不提交位移利用Kafka天然重试机制重新消费消息达到最大重试次数后不再重试主动投递至专属死信Topic独立消费者监听死信队列实现日志归档、异常告警、问题排查提供死信消息修复重发接口人工修正后重新进入业务队列消费。6.4.2 带重试机制的业务消费者通过消息自定义Header透传重试次数实现阶梯重试规避无限重试阻塞问题。// 全局重试配置常量 #define MAX_RETRY_TIMES 8 // 最大重试次数 #define Kafka_DLQ_TOPIC_SUFFIX _DLQ // 死信Topic后缀 // 获取消息Header中的重试次数 int GetMsgRetryTimes(RdKafka::Message msg) { // 首次消费无Header默认重试0次 if (!msg.headers()) return 0; std::string retry_str; RdKafka::Headers* headers msg.headers(); // 提取自定义重试次数字段 if (headers-get(retry_times, retry_str) RdKafka::ERR_NO_ERROR) { return std::stoi(retry_str); } return 0; } // 给消息添加重试次数Header void SetMsgRetryHeader(RdKafka::Producer* producer, RdKafka::Message msg, int retry_times) { RdKafka::Headers headers; headers.add(retry_times, std::to_string(retry_times)); producer-headers_set(headers); } // 带重试、死信机制的核心消费者 void KafkaRetryConsumer() { std::string errstr; RdKafka::Conf* conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); RdKafka::Conf* tconf RdKafka::Conf::create(RdKafka::CONF_TOPIC); // 集群配置 conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); conf-set(group.id, cpp_kafka_retry_consumer_group, errstr); conf-set(enable.auto.commit, false, errstr); // 关闭自动提交手动管控 conf-set(auto.offset.reset, earliest, errstr); RdKafka::KafkaConsumer* consumer RdKafka::KafkaConsumer::create(conf, tconf, errstr); if (!consumer) { std::cerr 重试消费者创建失败 errstr std::endl; return; } // 订阅业务Topic std::string biz_topic cpp_kafka_biz_topic; consumer-subscribe({biz_topic}); std::cout 带重试机制的业务消费者启动成功 std::endl; // 初始化死信投递生产者 RdKafka::Conf* dlq_conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); dlq_conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); RdKafka::Producer* dlq_producer RdKafka::Producer::create(dlq_conf, errstr); while (true) { RdKafka::Message* msg consumer-consume(1000); if (msg-err() ! RdKafka::ERR_NO_ERROR) { delete msg; continue; } // 获取当前消息重试次数 int cur_retry GetMsgRetryTimes(*msg); std::cout 消费消息当前重试次数 cur_retry 消息内容 (char*)msg-payload() std::endl; // 模拟业务异常数据库故障、接口超时、数据异常等 bool biz_failed true; if (!biz_failed) { // 业务处理成功手动提交位移 consumer-commitSync(*msg); std::cout 消息处理成功提交位移 std::endl; } else { // 业务处理失败判断是否达到最大重试次数 if (cur_retry MAX_RETRY_TIMES) { // 未达上限重试投递累加重试次数 cur_retry; SetMsgRetryHeader(dlq_producer, *msg, cur_retry); // 重新投递原Topic实现重试 dlq_producer-produce( biz_topic, msg-partition(), RdKafka::MSG_COPY, msg-payload(), msg-len(), msg-key(), msg-key_len(), nullptr ); dlq_producer-poll(0); std::cout 业务处理失败第 cur_retry 次重试 std::endl; } else { // 达到最大重试次数转入死信队列 std::string dlq_topic biz_topic Kafka_DLQ_TOPIC_SUFFIX; dlq_producer-produce( dlq_topic, RdKafka::PARTITION_UA, RdKafka::MSG_COPY, msg-payload(), msg-len(), msg-key(), msg-key_len(), nullptr ); dlq_producer-poll(0); std::cerr 重试次数耗尽消息转入死信队列 dlq_topic std::endl; // 放弃当前消息提交位移避免阻塞后续消费 consumer-commitSync(*msg); } } delete msg; } consumer-close(); delete consumer; delete dlq_producer; delete conf; delete tconf; delete dlq_conf; }6.4.3 死信队列独立消费者告警归档单独部署死信消费者不影响主业务链路实现死信消息监控、日志持久化、异常告警。void KafkaDlqConsumer() { std::string errstr; RdKafka::Conf* conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); RdKafka::Conf* tconf RdKafka::Conf::create(RdKafka::CONF_TOPIC); conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); conf-set(group.id, cpp_kafka_dlq_consumer_group, errstr); conf-set(enable.auto.commit, true, errstr); conf-set(auto.offset.reset, earliest, errstr); RdKafka::KafkaConsumer* consumer RdKafka::KafkaConsumer::create(conf, tconf, errstr); if (!consumer) { std::cerr 死信消费者创建失败 errstr std::endl; return; } // 订阅死信Topic std::string dlq_topic cpp_kafka_biz_topic Kafka_DLQ_TOPIC_SUFFIX; consumer-subscribe({dlq_topic}); std::cout 死信队列消费者启动成功监听Topic dlq_topic std::endl; while (true) { RdKafka::Message* msg consumer-consume(1000); if (msg-err() ! RdKafka::ERR_NO_ERROR) { delete msg; continue; } // 死信消息业务处理 std::cerr 【KAFKA死信消息】 std::endl; std::cerr 消息内容 (char*)msg-payload() std::endl; std::cerr 消息分区 msg-partition() 消息位移 msg-offset() std::endl; // 拓展业务 // 1. 死信消息入库归档留存溯源数据 // 2. 调用钉钉/企业微信接口发送异常告警 // 3. 统计死信数量监控服务异常率 delete msg; } consumer-close(); delete consumer; delete conf; delete tconf; }6.4.4 死信消息人工修复重发工具针对格式错误、瞬时故障导致的死信消息修复数据后重新投递至业务队列恢复业务流转。// 死信消息修复重发 bool KafkaDlqResend(const std::string biz_topic, const std::stringamp msg_body, const std::stringamp key ) { std::string errstr; RdKafka::Conf* conf RdKafka::Conf::create(RdKafka::CONF_GLOBAL); conf-set(bootstrap.servers, 127.0.0.1:9092, errstr); RdKafka::Producer* producer RdKafka::Producer::create(conf, errstr); if (!producer) { std::cerr 重发生产者创建失败 errstr std::endl; return false; } // 修复后重新投递业务Topic重置重试次数 RdKafka::ErrorCode ret producer-produce( biz_topic, RdKafka::PARTITION_UA, RdKafka::MSG_COPY, (void*)msg_body.c_str(), msg_body.size(), key.empty() ? nullptr : key.c_str(), key.size(), nullptr ); producer-flush(1000); delete producer; delete conf; if (ret RdKafka::ERR_NO_ERROR) { std::cout 死信消息修复重发成功 std::endl; return true; } else { std::cerr 死信消息重发失败 RdKafka::err2str(ret) std::endl; return false; } } // 重发测试示例 void DlqResendTest() { // 修复异常消息内容 std::string fix_msg 修复后的正常业务消息; KafkaDlqResend(cpp_kafka_biz_topic, fix_msg, biz_key_001); }6.4.5 重试死信机制生产规范重试次数配置普通业务8次重试核心金融业务可调整为12次搭配阶梯重试间隔死信Topic规范统一使用「业务Topic_DLQ」命名便于统一管理和监控禁止死信堆积必须配置死信告警每日巡检、定期清理归档严格手动提交重试、死信场景必须关闭自动提交杜绝消息丢失、重复消费消息溯源死信消息必须留存原始Key、时间、分区、位移信息方便问题排查。七、完整 Main 函数测试入口int main() { // 基础发送 KafkaSyncProduce(); KafkaAsyncProduce(); KafkaBatchProduce(); KafkaOrderProduce(); // 集群高可用测试 KafkaClusterProduce(); // 全局单例测试 KafkaSingletonTest(); // 基础消费者单独运行 // KafkaConsume(); // KafkaManualOffsetConsume(); // 重试死信队列测试单独部署运行 // KafkaRetryConsumer(); // KafkaDlqConsumer(); // DlqResendTest(); return 0; }八、业务场景精准选型8.1 同步消息适用场景核心业务数据、计费日志、关键事件上报需要确保消息 100% 投递成功的场景8.2 异步/批量消息适用场景用户行为埋点、操作日志、访问流量上报服务器监控指标、设备采集数据大数据实时流式计算数据源8.3 顺序消息适用场景订单状态流转、业务流程审批金融流水、账务变更记录IM 消息时序推送8.4 手动 Offset 适用场景数据落地数据库、文件写入等耗时业务不允许重复消费、不允许丢失消息的核心链路8.5 集群高可用单例客户端场景线上长期运行的 C 网关、采集服务、中间件服务高并发、高可用、不允许单点故障的生产环境九、生产最佳实践 避坑指南禁止频繁创建 Producer必须全局单例避免端口句柄泄漏、连接风暴高吞吐优先批量异步开启 batch.size linger.ms极大提升吞吐核心业务关闭自动提交手动 commit 位移杜绝消息丢失与重复消费生产必须集群部署多 broker 节点容错杜绝单点故障顺序消息必须指定 Key保证同一业务 ID 路由同一分区重试死信规范业务异常统一走自定义重试机制超次数强制转入死信杜绝消息丢失与无限重试堆积顺序消息必须指定 Key保证同一业务 ID 路由同一分区十、全文总结1. librdkafka 是 C 接入 Kafka 的工业级高性能 SDK适配所有高吞吐流式业务场景2. 日常开发可根据业务可靠性要求选择同步、异步、批量、有序四种发送模式3. 生产环境必须配置多节点集群全局单例生产者手动位移提交保障高可用与数据一致性4. Kafka 主打高吞吐流式数据处理是日志采集、大数据实时计算、海量埋点场景的首选中间件。互动提问你在 C 接入 Kafka 时遇到过消息堆积、乱序、重复消费、连接泄漏等问题吗欢迎评论区交流