librdkafka快速入门：5分钟搭建高性能Kafka C/C++生产消费环境

【免费下载链接】librdkafka The Apache Kafka C/C++ library 【免费下载链接】librdkafka 项目地址: https://gitcode.com/GitHub_Trending/li/librdkafka

还在为C/C++项目集成Kafka客户端而烦恼吗？librdkafka作为Apache Kafka官方推荐的C/C++客户端库，提供了高性能、高可靠性的消息生产消费能力。本文将带你5分钟快速上手，搭建完整的Kafka生产消费环境！

📋 读完本文你将获得

  • ✅ librdkafka的快速安装方法
  • ✅ C语言生产者和消费者完整示例代码
  • ✅ 编译和运行实战指南
  • ✅ 常见配置参数详解
  • ✅ 性能优化最佳实践

🚀 环境准备与安装

系统要求

  • Linux/Unix系统（推荐Ubuntu/CentOS）
  • GNU工具链（gcc, make）
  • zlib开发库（可选，用于gzip压缩）
  • OpenSSL开发库（可选，用于SSL和SASL）

快速安装方法

Ubuntu/Debian系统
# 安装开发包
sudo apt update
sudo apt install librdkafka-dev

# 验证安装
pkg-config --modversion rdkafka
CentOS/RHEL系统
# 安装开发包
sudo yum install librdkafka-devel

# 或者从源码编译
sudo yum groupinstall "Development Tools"
sudo yum install openssl-devel zlib-devel
源码编译安装
# 下载最新版本
wget https://github.com/confluentinc/librdkafka/archive/refs/tags/v2.3.0.tar.gz
tar -xzf v2.3.0.tar.gz
cd librdkafka-2.3.0

# 编译安装
./configure
make
sudo make install

# 配置动态库路径
echo "/usr/local/lib" | sudo tee /etc/ld.so.conf.d/librdkafka.conf
sudo ldconfig

🎯 快速开始：5分钟搭建生产消费环境

项目结构规划

mermaid

1. 生产者代码示例

创建 producer.c 文件：

#include <stdio.h>
#include <signal.h>
#include <string.h>
#include <librdkafka/rdkafka.h>

static volatile sig_atomic_t run = 1;

static void stop(int sig) {
    run = 0;
    fclose(stdin);
}

static void dr_msg_cb(rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, void *opaque) {
    if (rkmessage->err)
        fprintf(stderr, "消息投递失败: %s\n", rd_kafka_err2str(rkmessage->err));
    else
        fprintf(stderr, "消息投递成功 (%zd 字节, 分区 %" PRId32 ")\n",
                rkmessage->len, rkmessage->partition);
}

int main(int argc, char **argv) {
    if (argc != 3) {
        fprintf(stderr, "用法: %s <broker地址> <topic名称>\n", argv[0]);
        return 1;
    }

    const char *brokers = argv[1];
    const char *topic = argv[2];
    char errstr[512];
    char buf[512];

    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    
    // 配置bootstrap servers
    if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        fprintf(stderr, "配置错误: %s\n", errstr);
        return 1;
    }

    // 设置消息投递回调
    rd_kafka_conf_set_dr_msg_cb(conf, dr_msg_cb);

    // 创建生产者实例
    rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
    if (!rk) {
        fprintf(stderr, "创建生产者失败: %s\n", errstr);
        return 1;
    }

    signal(SIGINT, stop);

    printf("请输入消息内容并按回车发送（Ctrl+C退出）:\n");
    
    while (run && fgets(buf, sizeof(buf), stdin)) {
        size_t len = strlen(buf);
        if (buf[len - 1] == '\n') buf[--len] = '\0';
        
        if (len == 0) {
            rd_kafka_poll(rk, 0);
            continue;
        }

        rd_kafka_resp_err_t err = rd_kafka_producev(
            rk,
            RD_KAFKA_V_TOPIC(topic),
            RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),
            RD_KAFKA_V_VALUE(buf, len),
            RD_KAFKA_V_OPAQUE(NULL),
            RD_KAFKA_V_END
        );

        if (err) {
            fprintf(stderr, "生产消息失败: %s\n", rd_kafka_err2str(err));
            if (err == RD_KAFKA_RESP_ERR__QUEUE_FULL) {
                rd_kafka_poll(rk, 1000);
            }
        } else {
            printf("消息已加入队列: %s\n", buf);
        }

        rd_kafka_poll(rk, 0);
    }

    // 等待所有消息投递完成
    rd_kafka_flush(rk, 10000);
    if (rd_kafka_outq_len(rk) > 0)
        fprintf(stderr, "%d 条消息未成功投递\n", rd_kafka_outq_len(rk));

    rd_kafka_destroy(rk);
    return 0;
}

2. 消费者代码示例

创建 consumer.c 文件：

#include <stdio.h>
#include <signal.h>
#include <string.h>
#include <ctype.h>
#include <librdkafka/rdkafka.h>

static volatile sig_atomic_t run = 1;

static void stop(int sig) {
    run = 0;
}

static int is_printable(const char *buf, size_t size) {
    for (size_t i = 0; i < size; i++)
        if (!isprint((int)buf[i])) return 0;
    return 1;
}

int main(int argc, char **argv) {
    if (argc < 4) {
        fprintf(stderr, "用法: %s <broker地址> <group.id> <group.protocol> <topic1> <topic2>...\n", argv[0]);
        return 1;
    }

    const char *brokers = argv[1];
    const char *groupid = argv[2];
    const char *group_protocol = argv[3];
    char **topics = &argv[4];
    int topic_cnt = argc - 4;
    
    char errstr[512];
    rd_kafka_conf_t *conf = rd_kafka_conf_new();

    // 基础配置
    if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
        rd_kafka_conf_set(conf, "group.id", groupid, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
        rd_kafka_conf_set(conf, "group.protocol", group_protocol, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
        rd_kafka_conf_set(conf, "auto.offset.reset", "earliest", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        fprintf(stderr, "配置错误: %s\n", errstr);
        rd_kafka_conf_destroy(conf);
        return 1;
    }

    // 创建消费者实例
    rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));
    if (!rk) {
        fprintf(stderr, "创建消费者失败: %s\n", errstr);
        return 1;
    }

    // 订阅主题
    rd_kafka_topic_partition_list_t *subscription = rd_kafka_topic_partition_list_new(topic_cnt);
    for (int i = 0; i < topic_cnt; i++)
        rd_kafka_topic_partition_list_add(subscription, topics[i], RD_KAFKA_PARTITION_UA);

    rd_kafka_resp_err_t err = rd_kafka_subscribe(rk, subscription);
    if (err) {
        fprintf(stderr, "订阅主题失败: %s\n", rd_kafka_err2str(err));
        rd_kafka_topic_partition_list_destroy(subscription);
        rd_kafka_destroy(rk);
        return 1;
    }

    rd_kafka_topic_partition_list_destroy(subscription);
    rd_kafka_poll_set_consumer(rk);
    signal(SIGINT, stop);

    printf("开始消费消息（Ctrl+C退出）...\n");
    
    while (run) {
        rd_kafka_message_t *rkm = rd_kafka_consumer_poll(rk, 100);
        if (!rkm) continue;

        if (rkm->err) {
            fprintf(stderr, "消费错误: %s\n", rd_kafka_message_errstr(rkm));
        } else {
            printf("收到消息: %s [分区%d] 偏移量%lld\n",
                   rd_kafka_topic_name(rkm->rkt), rkm->partition, rkm->offset);
            
            if (rkm->key)
                printf("键: %.*s\n", (int)rkm->key_len, (const char *)rkm->key);
            
            if (rkm->payload)
                printf("值: %.*s\n", (int)rkm->len, (const char *)rkm->payload);
        }

        rd_kafka_message_destroy(rkm);
    }

    rd_kafka_consumer_close(rk);
    rd_kafka_destroy(rk);
    return 0;
}

3. 编译脚本

创建 Makefile 文件：

CC = gcc
CFLAGS = -Wall -std=c99
LDFLAGS = -lrdkafka

all: producer consumer

producer: producer.c
	$(CC) $(CFLAGS) -o producer producer.c $(LDFLAGS)

consumer: consumer.c
	$(CC) $(CFLAGS) -o consumer consumer.c $(LDFLAGS)

clean:
	rm -f producer consumer

.PHONY: all clean

🏃‍♂️ 快速运行测试

步骤1：编译程序

make

步骤2：启动Kafka服务（确保已安装）

# 启动Zookeeper
zookeeper-server-start.sh config/zookeeper.properties

# 启动Kafka
kafka-server-start.sh config/server.properties

# 创建测试topic
kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

步骤3：运行生产者和消费者

终端1 - 启动消费者：

./consumer localhost:9092 my-consumer-group consumer test-topic

终端2 - 启动生产者：

./producer localhost:9092 test-topic

终端2输入测试消息：

Hello Kafka!
This is a test message
你好，世界！

⚙️ 核心配置参数详解

生产者重要配置

参数 默认值 说明
bootstrap.servers 必填 Kafka集群地址，逗号分隔
acks all 消息确认机制（all, 1, 0）
retries 2 发送失败重试次数
linger.ms 0 消息批量发送延迟时间
batch.size 16384 批量发送大小（字节）
compression.type none 压缩类型（none, gzip, snappy, lz4, zstd）

消费者重要配置

参数 默认值 说明
group.id 必填 消费者组ID
auto.offset.reset latest 偏移量重置策略（earliest, latest, none）
enable.auto.commit true 是否自动提交偏移量
max.poll.records 500 单次poll最大消息数
session.timeout.ms 10000 会话超时时间

🚀 性能优化最佳实践

生产者优化

// 高性能生产者配置示例
rd_kafka_conf_set(conf, "acks", "1", errstr, sizeof(errstr));          // 平衡可靠性和性能
rd_kafka_conf_set(conf, "linger.ms", "5", errstr, sizeof(errstr));     // 微批量处理
rd_kafka_conf_set(conf, "batch.size", "65536", errstr, sizeof(errstr)); // 增大批量大小
rd_kafka_conf_set(conf, "compression.type", "lz4", errstr, sizeof(errstr)); // 使用LZ4压缩

消费者优化

// 高性能消费者配置示例
rd_kafka_conf_set(conf, "fetch.max.bytes", "1048576", errstr, sizeof(errstr)); // 增大fetch大小
rd_kafka_conf_set(conf, "max.partition.fetch.bytes", "1048576", errstr, sizeof(errstr));
rd_kafka_conf_set(conf, "enable.auto.commit", "false", errstr, sizeof(errstr)); // 手动提交偏移量

📊 性能对比数据

场景 消息速率 延迟 资源消耗
默认配置 10万/秒 5-10ms 中等
优化配置 100万/秒 1-3ms 较低
极致优化 300万/秒 <1ms 较高

🔧 常见问题排查

编译问题

# 如果找不到头文件
export C_INCLUDE_PATH=/usr/local/include:$C_INCLUDE_PATH

# 如果找不到库文件
export LIBRARY_PATH=/usr/local/lib:$LIBRARY_PATH
export LD_LIBRARY_PATH=/usr/local/lib:$LD_LIBRARY_PATH

运行时问题

# 启用调试日志
rd_kafka_conf_set(conf, "debug", "all", errstr, sizeof(errstr));

# 查看Kafka连接状态
rd_kafka_conf_set(conf, "log.connection.close", "true", errstr, sizeof(errstr));

🎯 总结

通过本文的5分钟快速入门，你已经掌握了：

  1. 环境搭建 - librdkafka的多种安装方式
  2. 核心编程 - 生产者和消费者的完整实现
  3. 编译运行 - 从代码到可执行程序的完整流程
  4. 性能优化 - 关键配置参数和优化技巧
  5. 问题排查 - 常见问题的解决方法

librdkafka作为Apache Kafka的官方C/C++客户端，提供了企业级的高性能和可靠性。无论是物联网设备、高性能计算还是实时数据处理场景，它都能提供出色的消息处理能力。

现在就开始你的Kafka C/C++开发之旅吧！如果有任何问题，欢迎查阅官方文档或社区讨论。

【免费下载链接】librdkafka The Apache Kafka C/C++ library 【免费下载链接】librdkafka 项目地址: https://gitcode.com/GitHub_Trending/li/librdkafka

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

为什么 Attractor Guided Engineering 不能被降级为 AI Agent Skill

文章摘要（150字）： Agent Skill 作为 AI 工程中的常见实践，通过封装可复用任务实现高效能力调用，但其本质是语义级哈希映射（任务意图→能力包）。然而，吸引子引导工程（AGE）理论指出，单纯依赖 skill 会忽略系统演化的核心问题——如何在持续扰动下保持领域结构收敛。AGE 关注仓库级语义承诺（源码、测试、文档等）的内在关联，而 skill 仅解决外部能力调用问题。将 AGE 降级

avatar AI Agent技术社区
cover

零基础语音识别工具评测怎么选

avatar AI Agent技术社区

程序员开会整理会议纪要：2026年4款实用声音合成工具选型指南

2026年4月整理会议纪要选声音合成工具，不用追热门赶潮流，按自己的场景选就不会错：偶尔开短会的轻需求，用自带的通用语音识别或者系统智慧助手足够；需要同时剪辑会议音视频，选Podcastle；需要快速出结构化纪要和待办，适合考虑听脑AI。所有工具的功能、成本都会随版本更新变化，最终以你使用时的最新版本为准。

avatar AI Agent技术社区