Librdkafka是c语言实现的apachekafka的高性能客户端，为生产和使用kafka提供高效可靠的客户端，并且提供了c++接口

 

性能：

Librdkafka 是一款专为现代硬件使用而设计的高性能库，它尝试将内存复制保持在最小，可以让用户决定是需要高吞吐量还是低延迟的服务，性能调优的两个最重要的配置是：

*batch.num.messages：在发送消息之前累积在本地队列中等待的消息的最小数量。

*queue.buffering.max.ms：等待batch.num.messages多长时间来填写到本地队列中。

 

使用：

源码中的rdkafka.h、CONFIGURATION.md有Librdkafka的API的说明

初始化：

应用程序需要实例化一个顶级对象rd_kafka_t作为基础容器，提供全局配置和共享状态，调用rd_kafka_new()创建。

还需要实例化一个或多个topics(`rd_kafka_topic_t`)来提供生产或消费，topic对象保存topic特定的配置，并在内部填充所有可用分区和leader brokers，通过调用`rd_kafka_topic_new()`创建。

`rd_kafka_t`和`rd_kafka_topic_t`自带一个可选的配置API，如果没有调用API，Librdkafka将会使用CONFIGURATION.md中的默认配置。

 

注意

1.应用程序可能会创建多个`rd_kafka_t`对象，并且它们不共享任何状态

2.一个`rd_kafka_topic_t`对象仅可以用于创建它的`rd_kafka_t`对象

 

配置

为了简化与Apache Kafka官方软件的集成，降低学习曲线，librdkafka实现了与Apache Kafka官方客户端相同的配置属性。

使用`rd_kafka_conf_set()` 和`rd_kafka_topic_conf_set()`在创建对象之前应用配置。

注意：

`rd_kafka.._conf_t`对象在传递给rd_kafka.._new()`之后不可重复使用，调用`rd_kafka.._new()`后，应用程序不需要free任何配置资源。

 

例子

 

 

  1. rd_kafka_conf_t*conf;

  2. char errstr[512];

  3.  

  4. conf = rd_kafka_conf_new();

  5. rd_kafka_conf_set(conf, "compression.codec","snappy", errstr, sizeof(errstr));

  6. rd_kafka_conf_set(conf, "batch.num.messages", "100",errstr, sizeof(errstr));

  7.  

  8. rd_kafka_new(RD_KAFKA_PRODUCER,conf);


 

 

线程和回调函数

librdkafka内部使用多个线程来充分利用硬件资源.

API是线程安全的，应用程序可以在任意时间调用其线程内的任意api函数.

poll-based的API用于向应用程序提供信号，应用程序定期调用` rd_kafka_poll() `,poll API将会调用如下的API:

*消息传递报告回调函数：消息传递成功或失败的信号，允许应用程序释放消息中使用的任何应用程序资源。

*错误回调函数：发出错误信号，这些错误通常具有信息性质，例如连接broker失败，应用程序通常不需要做任何处理，错误的类型通过` rd_kafka_resp_err_t `枚举值传递，包括远程的broke错误和本地错误。

 

可选回调不是通过poll触发的，可以通过任意线程调用：

*Logging callback ：允许应用程序输出librdkafka生成的日志消息

*partitioner callback：应用提供的消息分区器，可在任意时刻、任意线程中调用，对于相同的键，可以调用多次

 

Brokers

Librdkafka需要至少一个brokers的初始化list，称作` bootstrap brokers `,通过"metadata.broker.list"配置属性或`rd_kafka_brokers_add()`来指定，用来连接所有bootstrapbrokers，并查询每个元数据的信息，其中包含brokers、topic、partitions和它们在kafka cluster中的leaders的完整列表，

Brokers的名字被指定为"host[:port]"，端口可选（默认9092），host是主机名或ip地址，如果主机解析到多个地址，librdkafka将轮询每个尝试连接的地址，因此，可以使用包含所有brokers地址的DNS记录来提供可靠的bootstrap broker。

 

Producer API

使用`RD_KAFKA_PRODUCER`设置了`rd_kafka_t`对象，并设置了一个或多个`rd_kafka_topic_t`对象后，librdkafka已经准备好接收要发送给brokers的消息。

`rd_kafka_produce()`函数有如下参数：

*`rkt` - 需要produce的topic，之前通过`rd_kafka_topic_new()`函数创

*`partition` - 生产到的partition，如果设置为`RD_KAFKA_PARTITION_UA`(UnAssigned)，那么配置的分区函数将会用来选择目标分区。

*`msgflags` - 0，或者是：

         * `RD_KAFKA_MSG_F_COPY` - librdkafka会立刻生成payload的一份拷贝，当payload在非持久化内存中（例如堆）时使用。

         * `RD_KAFKA_MSG_F_FREE` - librdkafka使用完payload后，会使用`free(3)`将其释放。

这两个指标是互斥的，如果既不需要copy也不需要free，那么这两个指标都不需要设置。

 

如果`RD_KAFKA_MSG_F_COPY`没有设置，将不会执行数据的复制，librdkafka将会hold住payload的指针直到消息成功传输或传输失败。

当librdkafka完成消息的传递，使应用程序重新获得payload内存的所有权后，传递报告回调函数将会被调用

如果设置了`RD_KAFKA_MSG_F_FREE`，传递报告回调函数不能对payload进行free

*`payload`,`len` - 消息的payload

*`key`,`keylen` - 可以用来进行消息分区的消息键

         它将被传递到topic分区回调函数（如果存在的话），并在发送给broker的时候附加在消息上

*`msg_opaque` - 应用程序提供的一个可选的每条消息的不透明指针，在消息回调函数中提供，让应用程序引用一个特定的指针。

 

`rd_kafka_produce()`是一个非阻塞API,它会在内部队列中排列消息并立即返回。如果已排列的消息个数超过了"queue.buffering.max.messages"配置项，`rd_kafka_produce()`返回-1并将errno设置为`ENOBUFS`，从而提供了一种背压机制

 

Simple Consumer API

NOTE: 对于高级KafkaConsumer接口，查看rd_kafka_subscribe(rdkafka.h) 或者 KafkaConsumer (rdkafkacpp.h)。

使用`RD_KAFKA_CONSUMER`和`rd_kafka_topic_t`实例创建`rd_kafka_t`后，应用程序还必须通过调用`rd_kafka_consume_start()`来为给定的分区启动consumer。

 

`rd_kafka_consume_start()` 参数：

  * `rkt` - 需要消费的topic，之前通过`rd_kafka_topic_new()`创建。

  *`partition` - 从哪个分区消费

  *`offset` - 开始消费的消息offset，这可能是绝对消息偏移或两个特殊偏移之一：

         `RD_KAFKA_OFFSET_BEGINNING` ：从partition队列的起始位置开始消费（最老的message）

         `RD_KAFKA_OFFSET_END`：在下一个要生产到该partition上的消息处开始消费

         `RD_KAFKA_OFFSET_STORED`：使用存储的offset

 

一个topic+partition的consumer启动后，librdkafka将会尝试通过反复从broker获取批次消息以保持本地队列中保存"queued.min.messages"条消息，然后这个本地消息队列将会通过三个不同的consume API传递给应用程序：

*`rd_kafka_consume()` - consume单条消息

*`rd_kafka_consume_batch()` - consume单条或多条消息

*`rd_kafka_consume_callback()` - consume本地队列中的所有消息，并给每条消息调用一个回调函数

这三个API按照性能升序排列，`rd_kafka_consume()`最慢，`rd_kafka_consume_callback()`最快。

使用`rd_kafka_message_t`类型标识一条已消费的消息，其成员为：

*`err` - 发回到应用程序的错误信号，如果不为0，那么`payload`成员将被认为是一条错误消息，`err`是错误码(`rd_kafka_resp_err_t`)，如果为0，`payload`则包含消息数据。

*`rkt`,`partition` - 该消息的topic和partition

*`payload`,`len` - payload消息，或者是错误信息(err!=0)

*`key`,`key_len` - 生产者指定的可选消息key

*`offset` - Message offset

 

`payload`和`key`以及整个消息的内存，属于librdkafka，调用`rd_kafka_message_destroy()`后不可再次使用，librdkafka将为该消息集的所有消息payloads共享相同的消息集接收缓冲存储器，以避免过度复制，这意味着如果应用程序决定hang on单个rd_kafka_message_t，它将阻止从相同消息集中释放所有其他消息的备份内存。

 

当应用程序完成从topic+partition的消息消费后，需要调用`rd_kafka_consume_stop()`来停止这个consumer，这也将清除本地队列中的当前的消息。

 

Offset management

broker version >= 0.9.0结合使用高版本的KafkaConsumer接口，可实现基于Broker的offset管理（查看rdkafka.h或 rdkafkacpp.h）

还可以通过本地文件存储来实现Offset管理，通过如下的topic配置参数，offset被永久写在本地文件中：

  * `auto.commit.enable`

  * `auto.commit.interval.ms`

  * `offset.store.path`

  * `offset.store.sync.interval.ms`

 

目前还没有对ZooKeeper的偏移量管理的支持。

Consumer groups

当kafka broker 版本>= 0.9 ，librdkafka支持基于broker的consumer groups

 

Topics

Librdkafka支持自动创建topic，broker需要配置"auto.create.topics.enable=true"

 

++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

一.编译librdkafka

环境：Fedora 20，32位

依赖：pthreads（必选），zlib（可选），libssl-dev（可选），libsasl2-dev（可选）

先查看自己的linux上是否安装了pthreads，指令如下：

# locate libpthread

因为我之前安过了，所以可以直接编译librdkafka，没有安的下个pthreads的源码——configure、make、make install。

开始编译librdkafka，指令如下：

 

  1. # ./configure

  2. # make

  3. # make install

lib库会被默认安装到/usr/local/lib目录

头文件被默认安装到/usr/local/include/librdkafka目录

二.生产者

新建Qt控制台工程KafkaProducer，Pro文件如下：

 

  1. #-------------------------------------------------

  2. #

  3. # Project created by QtCreator 2018-03-27T19:45:09

  4. #

  5. #-------------------------------------------------

  6. QT -= gui core

  7.  

  8. TARGET = KafkaProducer

  9. CONFIG += console

  10. CONFIG -= app_bundle

  11.  

  12. TEMPLATE = app

  13.  

  14. SOURCES += main.cpp

  15. INCLUDEPATH += /usr/local/include/librdkafka

  16. LIBS += -L/usr/local/lib -lrdkafka

  17. LIBS += -L/usr/local/lib -lrdkafka++

main.cpp文件如下：

 

  1. #include <iostream>

  2. #include <string>

  3. #include <cstdlib>

  4. #include <cstdio>

  5. #include <csignal>

  6. #include <cstring>

  7.  

  8. #include <getopt.h>

  9.  

  10. #include "rdkafkacpp.h"

  11.  

  12. static bool run = true;

  13.  

  14. static void sigterm (int sig) {

  15. run = false;

  16. }

  17.  

  18. class ExampleDeliveryReportCb : public RdKafka::DeliveryReportCb {

  19. public:

  20. void dr_cb (RdKafka::Message &message) {

  21. std::cout << "Message delivery for (" << message.len() << " bytes): " <<

  22. message.errstr() << std::endl;

  23. if (message.key())

  24. std::cout << "Key: " << *(message.key()) << ";" << std::endl;

  25. }

  26. };

  27.  

  28. class ExampleEventCb : public RdKafka::EventCb {

  29. public:

  30. void event_cb (RdKafka::Event &event) {

  31. switch (event.type())

  32. {

  33. case RdKafka::Event::EVENT_ERROR:

  34. std::cerr << "ERROR (" << RdKafka::err2str(event.err()) << "): " <<

  35. event.str() << std::endl;

  36. if (event.err() == RdKafka::ERR__ALL_BROKERS_DOWN)

  37. run = false;

  38. break;

  39.  

  40. case RdKafka::Event::EVENT_STATS:

  41. std::cerr << "\"STATS\": " << event.str() << std::endl;

  42. break;

  43.  

  44. case RdKafka::Event::EVENT_LOG:

  45. fprintf(stderr, "LOG-%i-%s: %s\n",

  46. event.severity(), event.fac().c_str(), event.str().c_str());

  47. break;

  48.  

  49. default:

  50. std::cerr << "EVENT " << event.type() <<

  51. " (" << RdKafka::err2str(event.err()) << "): " <<

  52. event.str() << std::endl;

  53. break;

  54. }

  55. }

  56. };

  57.  

  58. int main ()

  59. {

  60. std::string brokers = "localhost";

  61. std::string errstr;

  62. std::string topic_str="test";

  63. int32_t partition = RdKafka::Topic::PARTITION_UA;

  64.  

  65. RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

  66. RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

  67.  

  68. conf->set("bootstrap.servers", brokers, errstr);

  69.  

  70. ExampleEventCb ex_event_cb;

  71. conf->set("event_cb", &ex_event_cb, errstr);

  72.  

  73. signal(SIGINT, sigterm);

  74. signal(SIGTERM, sigterm);

  75.  

  76. ExampleDeliveryReportCb ex_dr_cb;

  77. conf->set("dr_cb", &ex_dr_cb, errstr);

  78.  

  79. RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);

  80. if (!producer) {

  81. std::cerr << "Failed to create producer: " << errstr << std::endl;

  82. exit(1);

  83. }

  84. std::cout << "% Created producer " << producer->name() << std::endl;

  85.  

  86. RdKafka::Topic *topic = RdKafka::Topic::create(producer, topic_str,

  87. tconf, errstr);

  88. if (!topic) {

  89. std::cerr << "Failed to create topic: " << errstr << std::endl;

  90. exit(1);

  91. }

  92.  

  93. for (std::string line; run && std::getline(std::cin, line);) {

  94. if (line.empty()) {

  95. producer->poll(0);

  96. continue;

  97. }

  98.  

  99. RdKafka::ErrorCode resp =

  100. producer->produce(topic, partition,

  101. RdKafka::Producer::RK_MSG_COPY /* Copy payload */,

  102. const_cast<char *>(line.c_str()), line.size(),

  103. NULL, NULL);

  104. if (resp != RdKafka::ERR_NO_ERROR)

  105. std::cerr << "% Produce failed: " <<

  106. RdKafka::err2str(resp) << std::endl;

  107. else

  108. std::cerr << "% Produced message (" << line.size() << " bytes)" <<

  109. std::endl;

  110.  

  111. producer->poll(0);

  112. }

  113.  

  114. run = true;

  115. // 退出前处理完输出队列中的消息

  116. while (run && producer->outq_len() > 0) {

  117. std::cerr << "Waiting for " << producer->outq_len() << std::endl;

  118. producer->poll(1000);

  119. }

  120.  

  121. delete conf;

  122. delete tconf;

  123. delete topic;

  124. delete producer;

  125.  

  126. RdKafka::wait_destroyed(5000);

  127.  

  128. return 0;

  129. }

  130.  

三.消费者

新建Qt控制台工程KafkaConsumer，Pro文件如下：

 

  1. #-------------------------------------------------

  2. #

  3. # Project created by QtCreator 2018-03-28T16:27:54

  4. #

  5. #-------------------------------------------------

  6. QT -= gui core

  7.  

  8. TARGET = KafkaConsumer

  9. CONFIG += console

  10. CONFIG -= app_bundle

  11.  

  12. TEMPLATE = app

  13.  

  14. SOURCES += main.cpp

  15. INCLUDEPATH += /usr/local/include/librdkafka

  16. LIBS += -L/usr/local/lib -lrdkafka

  17. LIBS += -L/usr/local/lib -lrdkafka++

main.cpp文件如下：

 

  1. #include <iostream>

  2. #include <string>

  3. #include <cstdlib>

  4. #include <cstdio>

  5. #include <csignal>

  6. #include <cstring>

  7.  

  8. #include <sys/time.h>

  9. #include <getopt.h>

  10. #include <unistd.h>

  11.  

  12. #include "rdkafkacpp.h"

  13.  

  14. static bool run = true;

  15. static bool exit_eof = true;

  16. static int eof_cnt = 0;

  17. static int partition_cnt = 0;

  18. static int verbosity = 1;

  19. static long msg_cnt = 0;

  20. static int64_t msg_bytes = 0;

  21.  

  22. static void sigterm (int sig) {

  23. run = false;

  24. }

  25.  

  26. class ExampleEventCb : public RdKafka::EventCb {

  27. public:

  28. void event_cb (RdKafka::Event &event) {

  29. switch (event.type())

  30. {

  31. case RdKafka::Event::EVENT_ERROR:

  32. std::cerr << "ERROR (" << RdKafka::err2str(event.err()) << "): " <<

  33. event.str() << std::endl;

  34. if (event.err() == RdKafka::ERR__ALL_BROKERS_DOWN)

  35. run = false;

  36. break;

  37.  

  38. case RdKafka::Event::EVENT_STATS:

  39. std::cerr << "\"STATS\": " << event.str() << std::endl;

  40. break;

  41.  

  42. case RdKafka::Event::EVENT_LOG:

  43. fprintf(stderr, "LOG-%i-%s: %s\n",

  44. event.severity(), event.fac().c_str(), event.str().c_str());

  45. break;

  46.  

  47. case RdKafka::Event::EVENT_THROTTLE:

  48. std::cerr << "THROTTLED: " << event.throttle_time() << "ms by " <<

  49. event.broker_name() << " id " << (int)event.broker_id() << std::endl;

  50. break;

  51.  

  52. default:

  53. std::cerr << "EVENT " << event.type() <<

  54. " (" << RdKafka::err2str(event.err()) << "): " <<

  55. event.str() << std::endl;

  56. break;

  57. }

  58. }

  59. };

  60.  

  61. void msg_consume(RdKafka::Message* message, void* opaque) {

  62. switch (message->err()) {

  63. case RdKafka::ERR__TIMED_OUT:

  64. //std::cerr << "RdKafka::ERR__TIMED_OUT"<<std::endl;

  65. break;

  66.  

  67. case RdKafka::ERR_NO_ERROR:

  68. /* Real message */

  69. msg_cnt++;

  70. msg_bytes += message->len();

  71. if (verbosity >= 3)

  72. std::cerr << "Read msg at offset " << message->offset() << std::endl;

  73. RdKafka::MessageTimestamp ts;

  74. ts = message->timestamp();

  75. if (verbosity >= 2 &&

  76. ts.type != RdKafka::MessageTimestamp::MSG_TIMESTAMP_NOT_AVAILABLE) {

  77. std::string tsname = "?";

  78. if (ts.type == RdKafka::MessageTimestamp::MSG_TIMESTAMP_CREATE_TIME)

  79. tsname = "create time";

  80. else if (ts.type == RdKafka::MessageTimestamp::MSG_TIMESTAMP_LOG_APPEND_TIME)

  81. tsname = "log append time";

  82. std::cout << "Timestamp: " << tsname << " " << ts.timestamp << std::endl;

  83. }

  84. if (verbosity >= 2 && message->key()) {

  85. std::cout << "Key: " << *message->key() << std::endl;

  86. }

  87. if (verbosity >= 1) {

  88. printf("%.*s\n",

  89. static_cast<int>(message->len()),

  90. static_cast<const char *>(message->payload()));

  91. }

  92. break;

  93.  

  94. case RdKafka::ERR__PARTITION_EOF:

  95. /* Last message */

  96. if (exit_eof && ++eof_cnt == partition_cnt) {

  97. std::cerr << "%% EOF reached for all " << partition_cnt <<

  98. " partition(s)" << std::endl;

  99. run = false;

  100. }

  101. break;

  102.  

  103. case RdKafka::ERR__UNKNOWN_TOPIC:

  104. case RdKafka::ERR__UNKNOWN_PARTITION:

  105. std::cerr << "Consume failed: " << message->errstr() << std::endl;

  106. run = false;

  107. break;

  108.  

  109. default:

  110. /* Errors */

  111. std::cerr << "Consume failed: " << message->errstr() << std::endl;

  112. run = false;

  113. }

  114. }

  115.  

  116. class ExampleConsumeCb : public RdKafka::ConsumeCb {

  117. public:

  118. void consume_cb (RdKafka::Message &msg, void *opaque) {

  119. msg_consume(&msg, opaque);

  120. }

  121. };

  122.  

  123. int main () {

  124. std::string brokers = "localhost";

  125. std::string errstr;

  126. std::string topic_str="test";

  127. std::vector<std::string> topics;

  128. std::string group_id="101";

  129.  

  130. RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

  131. RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

  132. //group.id必须设置

  133. if (conf->set("group.id", group_id, errstr) != RdKafka::Conf::CONF_OK) {

  134. std::cerr << errstr << std::endl;

  135. exit(1);

  136. }

  137.  

  138. topics.push_back(topic_str);

  139. //bootstrap.servers可以替换为metadata.broker.list

  140. conf->set("bootstrap.servers", brokers, errstr);

  141.  

  142. ExampleConsumeCb ex_consume_cb;

  143. conf->set("consume_cb", &ex_consume_cb, errstr);

  144.  

  145. ExampleEventCb ex_event_cb;

  146. conf->set("event_cb", &ex_event_cb, errstr);

  147. conf->set("default_topic_conf", tconf, errstr);

  148.  

  149. signal(SIGINT, sigterm);

  150. signal(SIGTERM, sigterm);

  151.  

  152. RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);

  153. if (!consumer) {

  154. std::cerr << "Failed to create consumer: " << errstr << std::endl;

  155. exit(1);

  156. }

  157. std::cout << "% Created consumer " << consumer->name() << std::endl;

  158.  

  159. RdKafka::ErrorCode err = consumer->subscribe(topics);

  160. if (err) {

  161. std::cerr << "Failed to subscribe to " << topics.size() << " topics: "

  162. << RdKafka::err2str(err) << std::endl;

  163. exit(1);

  164. }

  165.  

  166. while (run) {

  167. //5000毫秒未订阅到消息，触发RdKafka::ERR__TIMED_OUT

  168. RdKafka::Message *msg = consumer->consume(5000);

  169. msg_consume(msg, NULL);

  170. delete msg;

  171. }

  172.  

  173. consumer->close();

  174.  

  175. delete conf;

  176. delete tconf;

  177. delete consumer;

  178.  

  179. std::cerr << "% Consumed " << msg_cnt << " messages ("

  180. << msg_bytes << " bytes)" << std::endl;

  181.  

  182. //应用退出之前等待rdkafka清理资源

  183. RdKafka::wait_destroyed(5000);

  184.  

  185. return 0;

  186. }

四.测试

先启动zookeeper服务和kafka服务，详见：kafka的编译和使用，然后再启动生产者和消费者。

生产者循环等待用户输入，输入后回车，消息就发布出去了，此时消费者显示订阅到的内容。

转载地址：

https://blog.csdn.net/lijinqi1987/article/details/76571757

https://blog.csdn.net/caoshangpa/article/details/79786100

 

# kafka
Logo
Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一） 概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

avatar Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

avatar Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...

avatar Kafka开源项目指南