Работа с Apache Kafka в C++: библиотеки и примеры

Работа с Apache Kafka в C++: библиотеки, настройка и примеры

Apache Kafka — это распределённая платформа для потоковой обработки данных, широко применяемая в микросервисной архитектуре, системах логирования и аналитики. Для интеграции с C++ существует несколько зрелых библиотек, из которых наиболее популярны librdkafka (низкоуровневая) и cppkafka (обёртка на C++). В этой статье мы подробно разберём обе библиотеки, покажем примеры кода продюсера и консюмера, а также дадим рекомендации по выбору.

1. librdkafka — основа для работы с Kafka на C

librdkafka — это высокопроизводительная C-библиотека от Confluent, реализующая протокол Apache Kafka. Она поддерживает все ключевые функции: асинхронную отправку сообщений, группы потребителей, exactly-once семантику (через транзакции) и 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.git
cd librdkafka
./configure --prefix=/usr/local
make
sudo make install

Пример продюсера на C с использованием librdkafka

Простой продюсер, отправляющий сообщение в топик:

#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 << "Failed to create producer: " << errstr << std::endl;
        return 1;
    }

    std::string topic_name = "test-topic";
    std::string message = "Hello from C++ and 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 << "Failed to produce message: " << rd_kafka_err2str(err) << std::endl;
    } else {
        std::cout << "Message sent successfully!" << std::endl;
    }

    // Ожидание доставки и очистка
    rd_kafka_flush(rk, 10000);
    rd_kafka_destroy(rk);
    return 0;
}

Пример консюмера на C с librdkafka

Простой потребитель, читающий сообщения из топика:

#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 << "Failed to create consumer: " << 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 << "Waiting for messages..." << std::endl;
    while (true) {
        rd_kafka_message_t *msg = rd_kafka_consumer_poll(rk, 1000);
        if (msg) {
            if (msg->err) {
                std::cerr << "Consumer error: " << rd_kafka_err2str(msg->err) <&