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

Библиотека librdkafka: мощный инструмент для работы с Kafka в C++

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

Основные возможности librdkafka

• Поддержка всех ключевых операций: отправка сообщений (producer) и чтение (consumer) с управлением смещениями (offset).
• Работа с топиками, партициями и группами потребителей.
• Нативная поддержка сжатия (gzip, snappy, lz4, zstd).
• SSL/TLS шифрование и аутентификация SASL (PLAIN, SCRAM, GSSAPI, OAUTHBEARER).
• Idempotent producer и транзакции (начиная с Kafka 0.11).
• Асинхронная обработка с колбэками и низкой задержкой.
• Тонкая настройка через конфигурационные свойства (аналоги Java-клиенту).

Установка и подключение

Установить librdkafka можно через пакетный менеджер или собрать из исходников. Для Linux (Ubuntu/Debian) команда:

sudo apt-get install librdkafka-dev

Для macOS через Homebrew:

brew install librdkafka

После установки подключите заголовочный файл и библиотеку при компиляции:

g++ -o my_app my_app.cpp -lrdkafka -lpthread -lz

Пример: создание продюсера (Producer)

Продюсер отправляет сообщения в указанный топик. Рассмотрим минимальный рабочий пример:

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

int main() {
    std::string errstr;
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

    // Настройка брокера
    conf->set("bootstrap.servers", "localhost:9092", errstr);

    // Создание продюсера
    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
    if (!producer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        return 1;
    }
    delete conf;

    // Отправка сообщения
    std::string topic_name = "test-topic";
    std::string payload = "Hello from C++ Kafka!";
    RdKafka::ErrorCode resp = producer->produce(
        topic_name,
        RdKafka::Topic::PARTITION_UA,
        RdKafka::Producer::RK_MSG_COPY,
        const_cast<char*>(payload.c_str()),
        payload.size(),
        nullptr, 0, 0, nullptr
    );

    if (resp != RdKafka::ERR_NO_ERROR) {
        std::cerr << "Produce failed: " << RdKafka::err2str(resp) << std::endl;
    } else {
        std::cout << "Message sent successfully" << std::endl;
    }

    // Ожидание доставки
    producer->flush(10000);
    delete producer;
    return 0;
}

В этом примере мы создаём конфигурацию, указываем адрес брокера, создаём экземпляр Producer и отправляем строковое сообщение в топик test-topic. Метод flush гарантирует, что все сообщения будут отправлены до завершения программы.

Пример: создание консюмера (Consumer)

Консюмер читает сообщения из топика в составе группы потребителей. Пример с ручным управлением смещениями:

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

class MyConsumerCb : public RdKafka::ConsumeCb {
public:
    void consume_cb(RdKafka::Message &msg, void *opaque) override {
        if (msg.err()) {
            if (msg.err() != RdKafka::ERR__TIMED_OUT) {
                std::cerr << "Error: " << msg.errstr() << std::endl;
            }
            return;
        }
        std::cout << "Received: " 
                  << static_cast<char*>(msg.payload()) 
                  << " (offset: " << msg.offset() << ")" <