将序列化的thrift结构序列化为c中的kafka++

nzkunb0c  于 2021-06-07  发布在  Kafka
关注(0)|答案(1)|浏览(382)

我有一套 structs 定义于 Thrift 例如:

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

我需要做以下几件事 C++ :
(a) 序列化的示例 Foo 转换为与节约兼容的字节(使用 Binary 或者 Compact 节俭协议)
(b) 将字节序列化示例发送到 Kafka 主题
问题
我如何发送 Thrift 序列化示例到 Kafka 集群?
提前谢谢

1rhkuytd

1rhkuytd1#

找到了我自己问题的答案。

序列化

下面的代码片段演示了如何序列化的示例 FooThrift -兼容字节(使用节俭 Compact 协议)。为了使用 Binary 协议,替换 TCompactProtocolTBinaryProtocol .


# 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集群

下面的代码片段演示了如何将与节俭兼容的字节发送到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);

相关问题