文章源地址：https://github.com/edenhill/librdkafka/blob/master/INTRODUCTION.md

librdkafka 是Apache  Kafka  客户端C语言的高性能实现， 能够提供可靠并且表现优秀的客户端，同时它也提供比较初级的C++界面。

Contents

本文主要包含以下章节：

一、性能

-性能指标

-高吞吐量

-低延迟

-压缩

二、消息可靠性

三、用法

-文档介绍

-初始化

-配置

-线程和回调函数

-brokers

-producer API

-consumer API

四、其他

-测试细节

一、性能

librdkafka库是多线程的，给现在硬件系统设计，并尽力实现最小的内存拷贝。生产或消费的消息的payload在传输中没有拷贝，同时消息的尺寸没有任何限制。

“你可能需要高吞吐率或者低延迟，但你可以拥有这两个性能”。

librdkafka 允许你实现高吞吐率或者低延迟，这是通过可配置的属性设置实现的。

性能指标中最重要的两个配置属性是：

-batch.num.messages：在发送消息序列之前，本地消息队列中需要积累的最小消息数。

-queue.buffering.max.ms：等待batch.num.messags在本地队列中实现的时间长度。

1、性能指标

后面的测试都是使用以下性能配置指标：

-intel  Quad  Core i7  at 3.4GHz， 8GB 内存

-通过设置brokers 刷新的配置属性，采用简单的方式测试硬盘性能。

       log.flush.interval.messages=10000000

      log.flush.interval.ms=100000

-两个brokers和librdkafka都运行在同一台机器上

-每个topic有两个partitions

-每个broker只是一个partitions的leader

-使用子目录example下的rdkafka_performance进行测试

测试结果（注意，原文只有producer的测试）

Test1: 2 brokers, 2 partitions, required.acks=2, 100 byte messages: 850000 messages/second, 85 MB/second


Test2: 1 broker, 1 partition, required.acks=0, 100 byte messages: 710000 messages/second, 71 MB/second


Test3: 2 broker2, 2 partitions, required.acks=2, 100 byte messages, snappy compression: 300000 messages/second, 30 MB/second


Test4: 2 broker2, 2 partitions, required.acks=2, 100 byte messages, gzip compression: 230000 messages/second, 23 MB/second

注意：本文最后将描述详细的测试信息

注意：consumer的测试结果很快就会补上

2、高吞吐率

高吞吐率的关键是消息批处理实现--首先本地消息队列中会积累一定数量的消息，然后再将这些消息一块发送出去。这就减少了消息传递消耗并且减少了来回申请产生不良影响。

默认设置下： batch.num.messages=1000，queue.buffering.max.ms=1000,这有利于高吞吐率。默认设置使得librdkafka向broker发送累积消息之前，可以等待1000ms，消息最多累积1000条。这两个属性哪一个先满足，就停止消息累积并进行发送，无论另一个属性是否满足。

这些设置虽然是全局性的（在rd_kafka_conf_t结构中实现），但是却适用于每个top+partitions基本组成。

3、低延迟

当要求消息发送低延迟时，“queue.buffering.max.ms"应当符合producer-side延迟所允许的最大值。 queue.buffering.max.ms设置为0将使得消息尽可能快的发送。

3、压缩

Producer 消息压缩通过“compression.codec配置属性实现。

压缩是批处理本地队列中的消息的，批处理消息的数目越大，压缩率越高。本地批处理队列的容量取决于”batch.num.messages“”queue.buffering.max.ms"配置属性，这在上边高吞吐率章节讨论。

二、消息可靠性

消息可靠性是librdkafka重要指标----实际应用中可以通过两个特定设置（“reuqest.required.acks”和“message.send.max.retries”）保证消息可靠性。

如果topic配置属性"request.reuired.acks“（除了0之外的其他值，查看具体细节）设置用来等待brokers收到消息的确认回复，则librdkafka在收到所有预期acks之前会一直保存消息，这就可以很好的处理一下事件：

-brokers 连接失败

-topic leader发生变化

-brokers通知produce错误

这些都是librdkafka自动完成，具体应用中无需针对以上事件做任何处理。消息在收到失败反馈之前可以保有的时间为”message.send.max.retires"。

librdkafka使用回调函数应对不同发送报告，即应对不同的消息发送状态，它将在收到每条消息传送状态时调用响应的回调函数。

-如果error_code  非0，则消息发送失败，error_code指明失败原因（rd_kafka_resp_err_t enum）

-如果error_code 为0，则消息成功发送

更多细节需要查看Producer  API章节。

发送报告的回调函数是可选的。

三、用法

1、文档介绍

librdkafka API描述在rdkafka.h中，配置属性描述在CONFIGURATION.md。

2、初始化

实际应用中，需要创建一个top-level的对象 rd_kafka_t， 这个对象是基本的容器，它提供了全局性配置属性以及共享状态信息，它由rd_kafka_new()函数创建。

同时也需要创建一个或者多个topics对象rd_kafka_topic_t，给produer以及consumer使用。 topic对象具有topic特定的配置属性，同时还包含了所有可用partitions与leader  brokers映射关系。它通过调用rd_kafka_topic_new()函数创建。

这两个对象都包含了可配置的API。默认情况下将调用默认值，具体属性默认值描述在CONFIGURATION.md。

注意：实际应用中，可能会创建多个rd_kafka_t对象，它们并没有共享状态信息

注意：rd_kafka_topic_t对象只能由创建它的对象rd_kafka_t使用。

3、配置

为了简化与kafka的集成以及缩短学习曲线，librdkafka实现配置属性都可以在kafka官方客户端中找到。

在创建对象之前，需要使用rd_kafka_conf_set()以及rd_kafka_topic_conf_set()函数进行配置。

注意： rd_kafka.._conf_t对象们在rd_kafka.._new()函数使用过后是不能被再次使用的，而且在rd_kakfa.._new()函数调用之后，不需要释放配置资源。

例子：

rd_kafka_conf_t * conf;
char errstr[512];

conf = rd_kafka_conf_new();
rd_kafka_conf_set( conf, "compression.codec", "snappy", errstr, sizeof(errstr));
rd_kafka_conf_set( conf, "batch.num.messages", "100", errstr, sizeof(errstr));

rd_kafka_new(RD_KAFKA_PRODUCER,conf);

4、线程和回调函数

librdkafka 内部将会有多个线程，以充分利用硬件资源。API的实现是完全线程安全的，实际应用中可以在任何时候任何线程中调用任何API函数而不用担心线程安全。

一个以轮询为基础的API用来给实际应用提供信号反馈，实际应用应当按照固定时间间隔调用rd_kafka_poll()函数。这个轮询的API将会调用以下可的回调（都是可选的）：

-消息发送报告回调：报告消息发送失败。这将允许实际应用采取措施应对发送失败，并释放消息发送过程中占有的资源。

-错误回调：报告错误；错误一般是信息化方面的，例如连接broker失败，实际应用通常不需要采取任何措施。错误的数据类型是通过rd_kafka_resp_err_t enum类型数据，可以描述本地错误和远程broker错误。

不是poll函数引起的可选回调函数，可能是由任意线程引发的：

-logging 回调：实际应用中，用于发送librdkafka产生的log消息。

-partitioner 回调：实际应用提供消息的partitioner。partitioner可能被任何线程任何时候调用，它可能由于同一个key而被调用多次。Partitioner 函数有以下限制：

     一定不能调用rd_kafka_*()等函数

     一定不能阻塞或延长执行

      一定要返回一个0到partition_cnt-1之间的值，或者是在partitioning不能执行的时候返回特定RD_KAFKA_PARTITION_UA值，

5、Brokers

librdkafka 只需要一份最初的brokers列表（至少包含一个broker）。它将连接所有"metadata.broker.list"或者是rd_kafka_brokers_add()函数添加的brokers，然后向每个brokers申请一些元数据信息：包含brokers的完整列表、topic、partitions以及它们在Kafka 集群中的leaders broker信息。

Brokers名字的形式为：host：port； 其中port是可选的，默认是9092，host是任何一个可以解析的hostname或者ipv4或者ipv6地址。如果host是多个地址，librdkafka将会在每一次连接尝试中循环连接这些地址。包含所有broker 地址的DNS记录可以用来提供可靠的bootstrap broker。

6、Producer  API

在使用RD_KAFKA_PRODUCER设置完rd_kafka_t对象后，就可以创建一个或者多个rd_kafka_topic_t对象了，用来接受信息或者发送信息。

rd_kafka_produce()函数需要以下参数：

-rkt： topic， 由前面rd_kafka_topic_new()函数创建

-partition：partition，如果为RD_KAFKA_PARTITION_UA，则配置的partitioner函数将会选择目标partition

-payload，len： 消息主体

-msgflags：0或者以下数值之一：

                   RD_KAFKA_MSG_F_COPY: librdkafka 在发送前先将消息拷贝下来，以防消息主体所在的缓存不是长久使用的，例如堆栈。

                   RD_KAFKA_MSG_F_FREE: librdkafka 在使用完消息后，将释放消息缓存。

                   这两个标志是互斥的，只能设置一个，用来表示是拷贝还是释放。

                    如果没有设置RD_KAFKA_MSG_F_COPY标志，则没有数据拷贝，librdkafka将会占有消息payload指针直到消息发送完毕或者失败。发送报告回调函数将会在librdkafka使实际调用重新获得payload缓存控制权的时候被调用。在RD_KAFKA_MSG_F_FREE设置的时候，实际调用一定不能在发送报告回调函数中释放payload。

-key，keylen： 可选参数，消息关键字，可以用来分区。它将会传递到topic partitioner回调中，如果存在，则会添加到发向broker的消息中。

-msg_opaque：可选参数，每条消息的透明度指针，由消息传送回调所提供，使应用参考特定的消息。

rd_kafka_produce()是非阻塞的API， 它将使消息存储在内部队列中并立即返回。如果入队的消息数目超过了配置的“queue.buffering.max.messages"属性，则rd_kafka_produce()函数将会返回-1并将errno设置为ENOBUFS， 这样就提供了应对压力的机制。

注意：examples/rdkafka_performance.c提供了producer的实现。

7、Consumer  API

consumer API要比producer  API多一些状态。 在使用RD_KAFKA_CONSUMER类型创建rd_kafka_t 对象，然后创建rd_kakfa_topic_t对象之后，实际应用中必须调用rd_kafka_consumer_start()函数启动对给定partition的consumer。

rd_kafka_consume_start()函数的参数：

-rkt： 进行consume的topic， 由前面rd_kafka_topic_new()创建

-partition：进行consume的partition

-offset：开始consume的消息偏移。这个偏移可能是一个绝对消息偏移，或者是RD_KAKFA_OFFSET_STORED来使用存储的offset，也可能是两个特定偏移之一：RD_KAFKA_OFFSET_BEGINNING，从partition消息队列的开始进行consume；RD_KAFKA_OFFSET_END：从partition中的将要produce的下一条信息开始（忽略即当前所有的消息）。

在topic+partition的consumer启动之后，librdkafka将尝试使本地消息队列中的消息数目保持在queued.min.messages，一方反复的从broker获取消息。

本地消息队列将通过以下三种不同的consum  APIs进行consume：

-rd_kafka_consume()：每次consume一条消息

-rd_kafka_consume_batch()：批处理consume，一条或多条

-rd_kafka_consume_callback()：consume本地消息队列中的所有消息，并调用回调函数处理每条消息

上述三种方式按照性能排列的，rd_kafka_consume()是最慢的，rd_kafka_consume_callback()最快。不同的需求可以选择不同的实现方式。

一条consumed消息，由每一个consume函数提供或返回，具体是由rd_kafka_messag_t类型对象保存。

rd_kafka_message_t对象成员：

-err：错误返回值。非0值表示出现错误，err是rd_kafka_resp_err_t类型数据。如果是0则表示进行了适当的消息抓取，并且payload中包含了message。

-rkt，partition：topic和partition信息

-payload，len：消息的payload数据或者错误的消息（err！=0）

-key，key_len：可选参数，主要是用来获取特定的消息。

-offset：消息的偏移地址

payload，key和消息一样，都是属于librdkafka，在rd_kafka_message_destroy()函数调用之后就不能再使用了。librdkafka将会使用相同的消息集接收缓存来存放消息消息集的playloads，这就避免过度拷贝，即意味着如果实际应用决定挂起某个单独的rd_kafka_message_t对象，这将会阻碍后面的缓存释放。

当实际应用完成consume消息，则应该调用rd_kafka_consume_stop()函数停止consumer。这将消除本地队列中中任何消息。

注意：examples/rdkafka_performance.c实现了consumer。

8、offset 管理

Offset管理可以通过本地offset保存文件完成，offset将会周期性的写入每个topic+partition的配置属性：

-auto.commit.enable

-auto.commit.interval.ms

-offset.store.path

-offset.store.sync.interval.ms

当前ZooKeeper还不支持offset管理。

9、Consumer  groups

当前还不支持consumer  groups， librdkafka consumer  API只编译了官方scala 简单版的Consumer。只有librdkafka能够支持这项应用，你才能拥有你的消费组。

10、Topics

Topic自动创建

topic自动创建是支持的。brokers需要使用”auto.create.topics.enable=true“进行配置。

四、其他：

测试细节：

Test1: Produce to two brokers, two partitions, required.acks=2, 100 byte messages

Each broker is leader for one of the two partitions. The random partitioner is used (default) and each broker and partition is assigned approximately 250000 messages each.

Command:

# examples/rdkafka_performance -P -t test2 -s 100 -c 500000 -m "_____________Test1:TwoBrokers:500kmsgs:100bytes" -S 1 -a 2
....
% 500000 messages and 50000000 bytes sent in 587ms: 851531 msgs/s and 85.15 Mb/s, 0 messages failed, no compression

Result:

Message transfer rate is approximately 850000 messages per second, 85 megabytes per second.

Test2: Produce to one broker, one partition, required.acks=0, 100 byte messages

Command:

# examples/rdkafka_performance -P -t test2 -s 100 -c 500000 -m "_____________Test2:OneBrokers:500kmsgs:100bytes" -S 1 -a 0 -p 1
....
% 500000 messages and 50000000 bytes sent in 698ms: 715994 msgs/s and 71.60 Mb/s, 0 messages failed, no compression

Result:

Message transfer rate is approximately 710000 messages per second, 71 megabytes per second.

Test3: Produce to two brokers, two partitions, required.acks=2, 100 byte messages, snappy compression

Command:

# examples/rdkafka_performance -P -t test2 -s 100 -c 500000 -m "_____________Test3:TwoBrokers:500kmsgs:100bytes:snappy" -S 1 -a 2 -z snappy
....
% 500000 messages and 50000000 bytes sent in 1672ms: 298915 msgs/s and 29.89 Mb/s, 0 messages failed, snappy compression

Result:

Message transfer rate is approximately 300000 messages per second, 30 megabytes per second.

Test4: Produce to two brokers, two partitions, required.acks=2, 100 byte messages, gzip compression

Command:

# examples/rdkafka_performance -P -t test2 -s 100 -c 500000 -m "_____________Test3:TwoBrokers:500kmsgs:100bytes:gzip" -S 1 -a 2 -z gzip
....
% 500000 messages and 50000000 bytes sent in 2111ms: 236812 msgs/s and 23.68 Mb/s, 0 messages failed, gzip compression

Result:

Message transfer rate is approximately 230000 messages per second, 23 megabytes per second.

Introduction to librdkafka - the Apache Kafka C/C++ client library 翻译