modern-cpp-kafka для C++. Решаем проблемы владения и сериализации

Доброго времени суток! Эта статья написана для тех, кто в общих чертах знаком с тем, что такое и для чего используется Apache Kafka, кто такие Producer и Consumer и как они друг с другом работают. Целью этой статьи является показать способ использования библиотеки modern-cpp-kafka для работы с Apache Kafka на современном C++. В общих чертах с темой можно ознакомится, например, здесь (отсюда же взяты скриншоты), а в этой статье будет рассмотрено решение проблем с владением и (де)сериализацией наиболее простым способом.

Идея написания этого небольшого руководства появилась у меня, когда я начал изучать одну из самых популярных библиотек для работы с Apache Kafka — а именно modern-cpp-kafka. Она основана в виде оболочки над старой доброй реализацией для C — lirdbkaka. Версия для C++ подкупила меня тем, что она предоставляет довольно удобный интерфейс для работы, использует современные и актуальные семантики языка вроде RAII и шаблонов, а также, как утверждает создатель, отсутствие оверхеда и самую быструю скорость при работе с данными размерами в пределах 256 B ~ 2 KB.

Однако, как водится, ничто не идеально, а именно — владение данными, сериализация и десериализация, которые реализованы что в librdkafka, что в modern-cpp-kafka примерно никак. Для того, чтобы разобраться, что именно не так, необходимо: рассмотреть механизм работы самой Kafka; ознакомиться с реализацией функций отправки и приема в librdkafka и modern-cpp-kafka; понять, причем здесь владение; разработать способ сериализации и десериализации.

В конце статьи я приведу код собственного решения проблемы с сериализацией. Решение отчасти банальное и не годится для промышленного использования, но оно хорошо подойдет начинающим пользователям ввиду своей минималистичности и простоты.

Механизм работы с данными в Apache Kafka

Максимально кратко рассмотрим механизм добавления и извлечения данных в Apache Kafka. Данные доставляются, хранятся и извлекаются исключительно в двоичном представлении, что автоматически поднимает вопрос о том, как получить данные в нужном пользователю виде — то есть о сериализации и десериализации.

Как видно на скриншоте, при добавлении ключа и соответствующего ему значения требуется преобразовать их в двоичный вид (сериализовать), для чего используются сериализаторы как для ключа, так и значения. Эти задачи берет на себя Producer.

Ситуация с извлечением данных выглядит очень похоже.

На этом скриншоте показана схема работы Consumer-а, который занимается извлечением двоичных данных и преобразованием их в нужный тип (десериализацией).

Механизм работы с данными в librdkafka и modern-cpp-kafka

Библиотека librdkafka, будучи написанной на C, использует именно C-шный механизм так называемого стирания типов (type erasure) путем использования функции rd_kafka_produce, а также связки void-указателя и длины данных.

RD_EXPORT int rd_kafka_produce(rd_kafka_topic_t *rkt,                      int32_t partition,                      int msgflags,                      void *payload,                      size_t len,                      const void *key,                      size_t keylen,                      void *msg_opaque);

Критика такого подхода стара как сам язык C, однако это единственный доступный в C способ передавать любые данные, нужно лишь их сериализовать. Тем не менее, сама библиотека не предоставляет никаких инструментов для сериализации.

Не решает ее, увы, и библиотека modern-cpp-kafka. Рассмотрим функцию KafkaProducer::send для отправки сообщения:

inline void KafkaProducer::send(const producer::ProducerRecord& record,     // Структура с данными (ключ, значение, хедеры, топик, партишен и id)                     const producer::Callback&       deliveryCb, // Callback-функция                     SendOption                      option,     // Опции для отправки                     ActionWhileQueueIsFull          action);    // Опции действия на заполненность очереди (блокировать или нет)

Нам нужен первый тип первого параметра — ProducerRecord. Его структура (не включая методы) выглядит так:

/**  * A key/value pair to be sent to Kafka.  * This consists of a topic name to which the record is being sent, an optional partition number, and an optional key and value.  * Note: `ProducerRecord` would not take the ownership from the memory block of `Value`.  */ class ProducerRecord { public:     using Id  = std::uint64_t;      // конструкторы, геттеры, сеттеры  private:     Topic        _topic;     Partition    _partition;     Key          _key;     Value        _value;     Headers      _headers;     Optional<Id> _id; };

Типы Key и Value являются type alias-ами на тип ConstBuffer:

// Which is similar with `boost::const_buffer` (thus avoid the dependency towards `boost`) class ConstBuffer { public:     // побайтовое копирование указателя     explicit ConstBuffer(const void* data = nullptr, std::size_t size = 0): _data(data), _size(size) {}     const void* data()     const { return _data; }     std::size_t size()     const { return _size; }     std::string toString() const     {         // опустим реализацию     } private:     const void* _data;     std::size_t _size; };

Поля класса выглядят как-то знакомо, не правда ли? Да, это именно та связка void-указателя и длины, которая используется и в функции rd_kafka_produce.

Копирование указателей выполняется побайтово, из-за этого мы сталкиваемся с проблемой «висящего» указателя и, как следствие, use-after-free. Именно реализация класса ConstBuffer делает класс ProducerRecord невладеющим (и крайне неудобным в нетривиальных случаях), что и сказано прямиков в коментарии в описании класса (для удобства переведенного на русский):

Примечание: ProducerRecord не будет принимать права владения блоком памяти Value.

Как мы видим, что в librdkafka, что в modern-cpp-kafka используется невладеющий механизм для передачи или хранения параметров.

Функции Serialize и Deserialize будут описаны ниже, в соответствующем пункте, пока же нам важно знать, что Serialize возврашает std::vector<std::byte>.Уже на этапе формировании данных для отправки возникает проблема с потерей данных:

// Хотим получить готовый объект из сериализованных данных // но получаем "висящий" указатель - увы и ах! template <typename T> ConstBuffer ValueFromNonOwning(const T& value) {   const std::vector<byte> serialized = Serialize(value);    // глубокого копирования не происходит - "висящий" указатель   return ConstBuffer(serialized.data(), serialized.size()); }  // Не спасет тут и умный указатель: ни на локальную переменную serialized // т.к. данные из умного указателя будут просто скопированы по значению // вместо создания копии; ни на сам возвращаемый объект ConstBuffer // (по изначальной причине) - и снова увы и ах! template <typename T> ConstBuffer ValueFromNonOwningUniquePtr(const T& value) {   // выделяем память в куче (с shared_ptr та же история, ведь счетчик ссылок равен 0)   const auto serialized = std::make_unique<std::vector<std::byte>>(Serialize(value));    // проблема остается, потому что теперь разрушается сам умный указатель   return ConstBuffer(serialized->data(), serialized->size()); }

Владеющий ConstBuffer

Чтобы починить проблему с «висящим» указателем, мне пришла идея написать тот владеющий буфер, который будет гарантировать сохранение данных на протяжении всей жизни объекта. Так появился OwningBuffer:

// Владеющий аналог класса ConstBuffer class OwningBuffer {  public:   explicit OwningBuffer(const void* data = nullptr, const std::size_t size = 0) {     if (data && size > 0) {       m_rawData.assign(static_cast<const std::byte*>(data),                        static_cast<const std::byte*>(data) + size);     }   }    explicit OwningBuffer(const ConstBuffer& buffer)       : OwningBuffer(buffer.data(), buffer.size()) {}    // конструктор копирования для создания из сырых данных   explicit OwningBuffer(const std::vector<std::byte>& bytes) : m_rawData(bytes) {}      // конструктор перемещения для создания из сырых данных   explicit OwningBuffer(std::vector<std::byte>&& bytes) noexcept       : m_rawData(std::move(bytes)) {}    [[nodiscard]] const void* data() const { return m_rawData.data(); }   [[nodiscard]] std::size_t size() const { return m_rawData.size(); }    [[nodiscard]] std::string toString() const {     // так как создание ConstBuffer не содержит сложных операций,     // нет и оверхеда на создание временного объекта     const ConstBuffer buffer(m_rawData.data(), m_rawData.size());     return buffer.toString();   }    // получение всегда валидного объекта ConstBuffer   [[nodiscard]] ConstBuffer asConstBuffer() const {     return ConstBuffer(m_rawData.data(), m_rawData.size());   }   private:   // сырые данные объекта   std::vector<std::byte> m_rawData; };  // Такой вариант функции работает корректно, никаких "висящих" указателей template <typename T> OwningBuffer ValueFrom(const T& value) {   const auto serialized = Serialize(value);    return OwningBuffer(serialized); }

Интерфейс класса аналогичен интерфейсу ConstBuffer, но содержит некоторые дополнения в виде новых конструкторов для создания объекта из сырых данных, а также вспомогательного метода для получения валидного объекта ConstBuffer. Владеющий подход полностью решает проблему с «висящими» указателями и use-after-free, и теперь можно перейти к сериализации и десериализации.

Сериализация и десериализация

Теперь, решив проблему с владением, можно перейти к самому главному — сериализации и десериализации данных. Несмотря на обилие библиотек, которые предоставляют такую функциональность, я не буду их использовать, чтобы не перегружать статью лишними зависимостями и усложнять материал разбором доступных средств для сериализации. Поэтому мы будем использовать старый добрый std::byte из C++17 в еще более старом std::vector и reinterpret_cast.

#include <cstring> #include <stdexcept> #include <type_traits> #include <vector>  template <typename T> std::vector<std::byte> Serialize(const T& value) {   static_assert(std::is_trivially_copyable_v<T>, "Type must be trivially copyable");    const auto begin = reinterpret_cast<const std::byte*>(&value);   return {begin, begin + sizeof(T)}; }  template <typename T> T Deserialize(const std::vector<std::byte>& serializedData) {   static_assert(std::is_trivially_copyable_v<T>, "Type must be trivially copyable");   static_assert(std::is_default_constructible_v<T>, "Type must be default constructible");    // примечание: проверяется только размер объекта.   // никто не мешает записать в int32_t содержимое std::string размером 5 байт   if (serializedData.size() != sizeof(T)) {     throw std::runtime_error("Serialized data size does not match target type size");   }    T value;   std::memcpy(&value, serializedData.data(), sizeof(T));   return value; }  // специализация для std::string template <> inline std::vector<std::byte> Serialize<std::string>(const std::string& value) {   auto begin = reinterpret_cast<const std::byte*>(value.data());   return {begin, begin + value.size()}; }  // специализация для std::string template <> inline std::string Deserialize<std::string>(     const std::vector<std::byte>& serializedData) {   return {reinterpret_cast<const char*>(serializedData.data()), serializedData.size()}; }

В static_assert-ах, по большому счету, нет такой необходимости, ведь код и так и так не скомпилируется, если что-то пойдет не так, но всегда приятнее видеть красивые сообщения об ошибках, верно? Сеханизм сериализации банален: берем адрес объекта и прибавляем его размер, таким образом «охватывая» все содержимое объекта, на выходе получаем вектор байтов, то есть сырые данные объекта.

С десериализацией все немного сложнее (и не так гладко). Здесь мы используем функцию memcpy, чтобы заполнить созданный при помощи конструктора по умолчанию объект переданными сырыми байтами. Также нужно иметь в виду, что проверяются только размеры десериализуемого объекта, то есть можно записать объект, например, std::string в объект типа float и получить бессмыслицу.

Остается лишь создать обертки для сериализации и десериализации. В этом нам помогут функции ValueTo и ValueFrom:

// сериализация template <typename T> OwningBuffer ValueFrom(const T& value) {   const auto serialized = Serialize(value);    return OwningBuffer(serialized); }  // десериализация template <typename T> T ValueTo(const Value& value) {   if (!value.data()) {     throw std::runtime_error("Received empty value");   }    // преобразуем void* к const byte* и "захватываем" все значения   const std::vector<std::byte> serializedData(       static_cast<const std::byte*>(value.data()),       static_cast<const std::byte*>(value.data()) + value.size());    // возвращаем объект, полученный при помощи десериализации   return Deserialize<T>(serializedData); }

Демонстрация работы

В качестве примера рассмотрим классическую модель Single Producer — SIngle Consumer. Код Producer-а:

#include <kafka/KafkaProducer.h>  #include <iostream> #include <string>  #include "KafkaUtils.h"  using namespace kafka::clients::producer;  void SendValues(const kafka::Topic& topic, KafkaProducer& producer) {   // Prepare delivery callback   auto deliveryCb = [](const RecordMetadata& metadata, const kafka::Error& error) {     if (!error) {       std::cout << "Message delivered: " << metadata.toString() << std::endl;     } else {       std::cerr << "Message failed to be delivered: " << error.message() << std::endl;     }   };    {     std::cout << "Sending messages with FLOAT type" << std::endl;      constexpr float f = 2.34;      kafka::extensions::SendValue(producer, topic, kafka::NullKey,                                  kafka::extensions::ValueFrom(f).AsConstBuffer(),                                  deliveryCb);   }    {     std::cout << "Sending messages with STRINGS type" << std::endl;      const auto values = std::vector<std::string>{"amogus", "breakpoint", "cappa",                                                  "delta",  "extension",  "final"};      for (const auto& value : values) {       kafka::extensions::SendValue(producer, topic, kafka::NullKey,                                    kafka::extensions::ValueFrom(value).AsConstBuffer(),                                    deliveryCb);     }   } }  void DoProducerWork() {   // взято из конфигурации Docker Compose   const std::string brokers = "localhost:29092";   const kafka::Topic topic = "test-topic";    // Prepare the configuration   kafka::Properties props;   props.put("bootstrap.servers", brokers);    // Create a producer   KafkaProducer producer(props);    std::println(std::cout, "SendValues function");   SendValues(topic, producer); }  int main() {   try {     DoProducerWork();   } catch (const std::exception& e) {     std::cerr << "Error: " << e.what() << std::endl;     return 1;   }    return 0; } 

Код Consumer-а:

#include <KafkaUtils.h> #include <kafka/KafkaConsumer.h>  #include <csignal> #include <iostream> #include <string>  std::atomic_bool running = {true};  void StopRunning(int sig) {   if (sig != SIGINT) return;    if (running) {     running = false;   } else {     // Restore the signal handler, -- to avoid stuck with this handler     signal(SIGINT, SIG_IGN);  // NOLINT   } }  void DoConsumerWork() {   using namespace kafka;   using namespace kafka::clients::consumer;    const std::string brokers = "localhost:29092";   const Topic topic = "test-topic";    // Prepare the configuration   Properties props;   props.put("bootstrap.servers", brokers);    // Create a consumer instance   KafkaConsumer consumer(props);    // Subscribe to topics   consumer.subscribe({topic});    while (running) {     // Poll messages from Kafka brokers     for (const auto records = consumer.poll(std::chrono::milliseconds(100));          const auto& record : records) {       if (record.error()) {         std::cerr << record.toString() << std::endl;         continue;       }        std::cout << "Got a new message..." << std::endl;       std::cout << "    Topic    : " << record.topic() << std::endl;       std::cout << "    Partition: " << record.partition() << std::endl;       std::cout << "    Offset   : " << record.offset() << std::endl;       std::cout << "    Timestamp: " << record.timestamp().toString() << std::endl;       std::cout << "    Headers  : " << toString(record.headers()) << std::endl;        try {         const auto stringValue = kafka::extensions::ValueTo<std::string>(record.value());         std::cout << "    STRING [" << stringValue << "]" << std::endl;       } catch (const std::exception& e) {         std::cerr << "    Failed to deserialize as string: " << e.what() << std::endl;       }        try {         const auto floatValue = kafka::extensions::ValueTo<float>(record.value());         std::cout << "    FLOAT [" << floatValue << "]" << std::endl;       } catch (const std::exception& e) {         std::cerr << "    Failed to deserialize as float: " << e.what() << std::endl;       }     }   } }  int main() {   // Use Ctrl-C to terminate the program   signal(SIGINT, StopRunning);  // NOLINT    DoConsumerWork();    return 0; } 

Заключение

В этой статье я постарался совместить сразу несколько вещей: рассказать про Apache Kafka, про хранение данных внутри, про наличие определенных проблем и подходы к их решению. Написание текстов (помимо курсовых и дипломной работ) для меня в новинку, поэтому призываю каждого, кто прочитал, поделиться мыслями — с удовольствием почитаю и отвечу.

Демонстрацию сериализации и десериализации прямо в браузере можно посмотреть на сайте godbolt. Исходный код с настройкой Apache Kafka в Docker Compose находится в моем github-репозитории.