在 C++ 中使用 Apache Kafka：库与示例

在 C++ 中使用 Apache Kafka：库、配置与示例

Apache Kafka 是一个分布式流数据处理平台，广泛应用于微服务架构、日志系统和分析系统。为了与 C++ 集成，有多个成熟的库，其中最流行的是 librdkafka（底层库）和 cppkafka（C++ 封装）。本文将详细解析这两个库，展示生产者和消费者的代码示例，并提供选型建议。

1. librdkafka — 在 C 中操作 Kafka 的基础

librdkafka 是 Confluent 开发的高性能 C 语言库，实现了 Apache Kafka 协议。它支持所有关键功能：异步消息发送、消费者组、精确一次语义（通过事务）以及 SSL/SASL。该库用 C 编写，但可以直接在 C++ 中使用，或通过 C++ 封装使用。

安装 librdkafka

在 Ubuntu/Debian 上：

sudo apt-get install librdkafka-dev

在 macOS 上通过 Homebrew：

brew install librdkafka

从源码编译：

git clone https://github.com/confluentinc/librdkafka.gitcd librdkafka./configure --prefix=/usr/localmakesudo make install

使用 librdkafka 的 C 语言生产者示例

一个简单的生产者，向主题发送消息：

#include <librdkafka/rdkafka.h>#include <iostream>#include <string>#include <cstring>

int main() {    rd_kafka_t *rk;          // 生产者句柄    rd_kafka_conf_t *conf;   // 配置    char errstr[512];

    conf = rd_kafka_conf_new();    rd_kafka_conf_set(conf, "bootstrap.servers", "localhost:9092", errstr, sizeof(errstr));    rd_kafka_conf_set(conf, "acks", "all", errstr, sizeof(errstr));

    rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));    if (!rk) {        std::cerr << "创建生产者失败: " << errstr << std::endl;        return 1;    }

    std::string topic_name = "test-topic";    std::string message = "来自 C++ 和 librdkafka 的问候！";

    rd_kafka_resp_err_t err = rd_kafka_producev(        rk,        RD_KAFKA_V_TOPIC(topic_name.c_str()),        RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),        RD_KAFKA_V_VALUE((void*)message.c_str(), message.size()),        RD_KAFKA_V_END    );

    if (err) {        std::cerr << "发送消息失败: " << rd_kafka_err2str(err) << std::endl;    } else {        std::cout << "消息发送成功！" << std::endl;    }

    // 等待投递并清理    rd_kafka_flush(rk, 10000);    rd_kafka_destroy(rk);    return 0;}

使用 librdkafka 的 C 语言消费者示例

一个简单的消费者，从主题读取消息：

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

int main() {    rd_kafka_t *rk;    rd_kafka_conf_t *conf;    rd_kafka_topic_partition_list_t *topics;    char errstr[512];

    conf = rd_kafka_conf_new();    rd_kafka_conf_set(conf, "bootstrap.servers", "localhost:9092", errstr, sizeof(errstr));    rd_kafka_conf_set(conf, "group.id", "my-group", errstr, sizeof(errstr));    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) {        std::cerr << "创建消费者失败: " << errstr << std::endl;        return 1;    }

    rd_kafka_poll_set_consumer(rk);    topics = rd_kafka_topic_partition_list_new(1);    rd_kafka_topic_partition_list_add(topics, "test-topic", RD_KAFKA_PARTITION_UA);    rd_kafka_consumer_subscribe(rk, topics);

    std::cout << "等待消息..." <<