C语言操作Kafka

Kafka服务

Kafka的快速入门 文档很详细，基本上几步就可以搭建一个Kafka测试环境。

下载Kafka的二进制包，然后解压。

wget https://www.apache.org/dyn/closer.cgi?path=/kafka/4.0.0/kafka_2.13-4.0.0.tgz
tar -xzf kafka_2.13-4.0.0.tgz
cd kafka_2.13-4.0.0

生成集群ID，使用集群ID格式化存储

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties

启动Kafka Server。

$ bin/kafka-server-start.sh config/server.properties

启动之后，默认的Kafka Server会连续输出日志到控制台。后面的测试命令，需要在另外的终端执行。

创建主题

bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

测试生产

bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
>This is my first event
>This is my second event

测试消费

bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event

C语言的Kafka库：librdkafka

在C语言中操作Kafka，有不少库可以选择，其中就有librdkafka。

根据librdkafka在gitthub上的信息，它主要是由C语言开发，但是也提供了C++的支持。

当我们在Linux平台上安装librdkafka之后，通过pkg-config命令加库名rdkafka，可以获取编译相关的信息。

如在Fedora 42上：

pkg-config --cflags --libs rdkafka  
-DWITH_GZFILEOP -lrdkafka

需要注意的是，rdkafka.pc中记录的头文件路径是/usr/include，但是它的头文件都位于/usr/include/librdkafka。在C语言的源代码中需要写成：

#include <librdkafka/rdkafka.h>

在C++中则需要写成：

#include <librdkafka/rdkafkacpp.h>

但是C++的API与C语言的类似，后续不再提及C++的相关API，仅以C语言举例。

rdkafka API

rd_kafka_t

API的核心是结构rd_kafka_t。

它的创建和销毁是一对函数：

rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,  rd_kafka_conf_t *conf,  char *errstr,  size_t errstr_size);  void rd_kafka_destroy(rd_kafka_t *rk);

其中，rd_kafka_type_t是一个enum，分别为RD_KAFKA_PRODUCER与RD_KAFKA_CONSUMER，即生产者与消费者。

rd_kafka_conf_t

rd_kafka_conf_t是另一个结构，创建与销毁函数为：

rd_kafka_conf_t *rd_kafka_conf_new(void);void rd_kafka_conf_destroy(rd_kafka_conf_t *conf);

rd_kafka_conf_t创建之后，使用如下函数设置参数：

rd_kafka_conf_res_t rd_kafka_conf_set(rd_kafka_conf_t *conf,  const char *name,  const char *value,  char *errstr,  size_t errstr_size);

如：bootstrap.servers、sasl.username、sasl.password等。

rd_kafka_conf_t设置相关的函数返回值都是rd_kafka_conf_res_t。如果成功，值为RD_KAFKA_CONF_OK，即0。

如果要进行更详细的控制，还可以使用其它一些函数。

如：

void rd_kafka_conf_set_dr_msg_cb(  rd_kafka_conf_t *conf,  void (*dr_msg_cb)(rd_kafka_t *rk,  const rd_kafka_message_t *rkmessage,  void *opaque));

可以设置消息被处理之后的回调函数。其中，rkmessage是被处理的消息，opaque是使用rd_kafka_conf_set_opaque()设置的用户层指针。

证书设置

rd_kafka_conf_set_ssl_cert用于设置证书。

rd_kafka_conf_res_t  
rd_kafka_conf_set_ssl_cert(rd_kafka_conf_t *conf,  rd_kafka_cert_type_t cert_type,  rd_kafka_cert_enc_t cert_enc,  const void *buffer,  size_t size,  char *errstr,  size_t errstr_size);

其中，cert_type的值分别为：RD_KAFKA_CERT_PUBLIC_KEY、RD_KAFKA_CERT_PRIVATE_KEY与RD_KAFKA_CERT_CA，即公钥、私钥与CA。

而cert_enc的值分别为：RD_KAFKA_CERT_ENC_PKCS12、RD_KAFKA_CERT_ENC_DER与RD_KAFKA_CERT_ENC_PEM，即分别是三种证书格式。

另外还可以设置证书验证的回调函数：

rd_kafka_conf_res_t rd_kafka_conf_set_ssl_cert_verify_cb(  rd_kafka_conf_t *conf,  int (*ssl_cert_verify_cb)(rd_kafka_t *rk,  const char *broker_name,  int32_t broker_id,  int *x509_error,  int depth,  const char *buf,  size_t size,  char *errstr,  size_t errstr_size,  void *opaque));

如果成功，回调函数必须返回1。否则需要设置errstr为错误原因，并且返回0。

rd_kafka_message_t

处理Kafka另外一个重要的结构是rd_kafka_message_t，即消息。它的定义为：

typedef struct rd_kafka_message_s {  rd_kafka_resp_err_t err;rd_kafka_topic_t *rkt; int32_t partition;void *payload;size_t len;                        void *key;size_t key_len;      int64_t offset;void *_private; 
} rd_kafka_message_t;

最开头的err非常重要。每次使用rd_kafka_consume*族函数取得一条消息，以及在生产一条消息的回调函数中时，都要检查这个值，确定是否成功。

生产者

使用rd_kafka_t生产一条消息的函数为：

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);

成功返回0。

生成一批消息的函数为：

int rd_kafka_produce_batch(rd_kafka_topic_t *rkt,  int32_t partition,  int msgflags,  rd_kafka_message_t *rkmessages,  int message_cnt);

返回值为成功的消息数目。

等待所有消息处理完成的函数为：

rd_kafka_resp_err_t rd_kafka_flush(rd_kafka_t *rk, int timeout_ms);

或者放弃没有发送的消息：

rd_kafka_resp_err_t rd_kafka_purge(rd_kafka_t *rk, int purge_flags);

消费者

控制Kafka消费的函数有一对：

int rd_kafka_consume_start(rd_kafka_topic_t *rkt,  int32_t partition,  int64_t offset);int rd_kafka_consume_stop(rd_kafka_topic_t *rkt, int32_t partition);

注意这两个函数的第一个参数是rd_kafka_topic_t。

调用rd_kafka_consume_start()之后，kafka将开始把成批的消息放入本地的队列中，应用需要使用rd_kafka_consume()函数来消费。

rd_kafka_message_t *  
rd_kafka_consume(rd_kafka_topic_t *rkt, int32_t partition, int timeout_ms);

这个函数在timeout_ms毫秒以内，返回一条消息，或者NULL。返回的消息，必须使用rd_kafka_message_destroy()释放。

类似生产者，消费者也有批量处理的函数：

ssize_t rd_kafka_consume_batch(rd_kafka_topic_t *rkt,  int32_t partition,  int timeout_ms,  rd_kafka_message_t **rkmessages,  size_t rkmessages_size);