librdkafka 库的使用

使用librdkafka库可以轻松地开发Kafka客户端应用程序。以下是使用librdkafka库的一些基本步骤：

1. 安装librdkafka库：您可以通过下载和编译源代码或使用操作系统的软件包管理器来安装librdkafka库。

2. 创建Kafka生产者/消费者配置：您需要设置Kafka集群的连接配置和其他参数，例如主题名称、分区、消息序列化器等。

3. 创建Kafka生产者/消费者：使用librdkafka库提供的API创建Kafka生产者/消费者对象。

4. 发送/接收消息：使用Kafka生产者/消费者对象发送/接收消息。您可以使用同步或异步API发送/接收消息。

5. 处理错误：处理可能发生的错误，例如连接错误、发送错误、接收错误等。

以下是使用librdkafka库创建Kafka生产者/消费者的示例代码：

#include <stdio.h>
#include <string.h>
#include <librdkafka/rdkafka.h>

int main(int argc, char **argv) {
    rd_kafka_t *rk;         /* Producer instance handle */
    rd_kafka_conf_t *conf;  /* Temporary configuration object */
    char errstr[512];       /* librdkafka API error reporting buffer */

    /* Kafka broker configuration */
    char *brokers = "localhost:9092";
    char *topic = "test_topic";

    /* Create Kafka configuration object */
    conf = rd_kafka_conf_new();

    /* Set bootstrap broker(s) */
    if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        fprintf(stderr, "Failed to set broker: %s\n", errstr);
        rd_kafka_conf_destroy(conf);
        return 1;
    }

    /* Create producer instance */
    rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
    if (!rk) {
        fprintf(stderr, "Failed to create producer: %s\n", errstr);
        rd_kafka_conf_destroy(conf);
        return 1;
    }

    /* Create message object */
    rd_kafka_topic_t *rkt;
    rkt = rd_kafka_topic_new(rk, topic, NULL);

    /* Produce message */
    const char *message = "Hello, Kafka!";
    size_t len = strlen(message);
    rd_kafka_resp_err_t err;
    err = rd_kafka_produce(rkt, RD_PARTITION_UA, RD_MSG_F_COPY, (void *)message, len, NULL, 0, NULL);
    if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
        fprintf(stderr, "Failed to produce message: %s\n", rd_kafka_err2str(err));
    }

    /* Wait for message delivery */
    rd_kafka_poll(rk, 0);

    /* Destroy message object */
    rd_kafka_topic_destroy(rkt);

    /* Destroy producer instance */
    rd_kafka_destroy(rk);

    return 0;
}

这是一个简单的Kafka生产者示例代码，它将消息“Hello, Kafka!”发送到Kafka集群。您可以使用类似的方式创建Kafka消费者。