在 C++ 中将序列化的 Thrift 结构序列化为 Kafka

Serializing a serialized Thrift struct to Kafka in C++

本文关键字：序列化 结构 Kafka Thrift C++      更新时间：2023-10-16

我有一组structs在Thrift中定义，如下所示：

struct Foo {
  1: i32 a,
  2: i64 b
}

我需要在C++执行以下操作：

(a( 将Foo实例序列化为兼容 Thrift 的字节(使用 Binary 或 Compact Thrift 协议(

(b( 将字节序列化实例发送到Kafka主题

问题

如何将Thrift序列化实例发送到Kafka集群？

提前致谢

想出了我自己问题的答案。

序列化

下面的代码片段说明了如何将 Foo 实例序列化为Thrift兼容的字节(使用 Thrift Compact 协议(。要使用Binary协议，请将TCompactProtocol替换为 TBinaryProtocol 。

#include <thrift/transport/TBufferTransports.h>
#include <thrift/protocol/TCompactProtocol.h>
using apache::thrift::protocol::TCompactProtocol;
using apache::thrift::transport::TMemoryBuffer;
...
...
boost::shared_ptr<TMemoryBuffer> buffer(new TMemoryBuffer());
boost::shared_ptr<TCompactProtocol> protocol(new TCompactProtocol(buffer));
uint8_t **serialized_bytes = reinterpret_cast<uint8_t **>(malloc(sizeof(uint8_t *)));
uint32_t num_bytes = 0;
// 'foo' is an instance of Foo
foo->write(protocol.get());
buffer->getBuffer(serialized_bytes, &num_bytes);

发送到 Kafka 集群

以下代码片段演示如何将与 Thrift 兼容的字节发送到 Kafka 群集。

注意：下面使用的 kafka 客户端库是 librdkafka。

#include "rdkafkacpp.h"
std::string errstr;
// Create global configuration
RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
conf->set("metadata.broker.list", "localhost:9092", errstr);
conf->set("api.version.request", "true", errstr);
// Create kafka producer
RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);
// Create topic-specific configuration
RdKafka::Topic *topic = RdKafka::Topic::create(producer, "topic_name", nullptr, errstr);
auto partition = 1;
// Sending the serialized bytes to Kafka cluster
auto res = producer->produce(
    topic, partition,
    RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
    serialized_bytes, num_bytes,
    NULL, NULL);
  if (res != RdKafka::ERR_NO_ERROR) {
    std::cerr << "Failed to publish message" << RdKafka::err2str(res) << std::endl;
  } else {
    std::cout << "Published message of " << num_bytes << " bytes" << std::endl;
  }
producer->flush(10000);
相关文章：