librdkafka带鉴权认证访问kafka服务器

librdkafka简介

librdkafka是用c语言实现的一个高性能的kafka客户端,因为性能强大,开发者们基于librdkafka开发了各种语言的kafka客户端,比如librdkafkad(c++),, node-rdkafka(Node.js), confulent-kafka-python(Python)等。
librdkafka的高性能主要体现在其多线程的设计以及尽可能的降低内存拷贝。

librdkakfa API 简介

librdkafka github地址:https://github.com/edenhill/librdkafka ,
其中,C语言API可以参考src/rdkafka.h头文件,简要介绍几个关键的对象

  • rd_kafka_t: kafka客户端对象
  • rd_kafka_conf_t: kafka客户端配置对象
  • rd_kafka_topic_t: kafka topic对象

创建这几个对象所使用的函数:

  • rd_kafka_new()
  • rd_kafka_conf_new()
  • rd_kafka_topic_new()

librdkafka支持多种协议以控制kafka服务器的访问权限,如SASL_PALIN, PLAINTEXT, SASL_SSL等,在使用librdkafka时需要通过security.protocol参数指定协议类型,再辅以相应协议所需的其它参数完成权限认证。

如果使用SASL协议进行权限认证,需要对librdkafka添加SASL库依赖并重新编译。例如:在CentOS下安装如下依赖包:

yum -y install cyrus-sasl cyrus-sasl-devel

经过重新编译librdkafka后,进入examples目录下,执行

./rdkafka_example -X builtin.features

结果为:

 builtin.features = gzip,snappy,ssl,sasl,regex

可以看到librdkafka已经有了sasl特性,后续可以通过sasl协议进行访问认证。

producer 代码示例

初始化producer


int KafkaApi::init_producer(const std::string &brokers,
                            const std::string &username,
                            const std::string &password) {
  char errstr[512];
  /* Kafka configuration */
  if (NULL == conf_) {
    conf_ = rd_kafka_conf_new();
  }

  rd_kafka_conf_set(conf_, "queued.min.messages", "20", NULL, 0);
  rd_kafka_conf_set(conf_, "bootstrap.servers", brokers.c_str(), errstr,
                    sizeof(errstr));
  rd_kafka_conf_set(conf_, "security.protocol", "sasl_plaintext", errstr,
                    sizeof(errstr));
  rd_kafka_conf_set(conf_, "sasl.mechanisms", "PLAIN", errstr, sizeof(errstr));
  rd_kafka_conf_set(conf_, "sasl.username", username.c_str(), errstr,
                    sizeof(errstr));
  rd_kafka_conf_set(conf_, "sasl.password", password.c_str(), errstr,
                    sizeof(errstr));
  rd_kafka_conf_set(conf_, "api.version.request", "true", errstr,
                    sizeof(errstr));
  rd_kafka_conf_set_dr_msg_cb(conf_, dr_msg_cb_trampoline);

  /* Create Kafka handle */
  if (!(rk_ = rd_kafka_new(RD_KAFKA_PRODUCER, conf_, errstr, sizeof(errstr)))) {
    fprintf(stderr, "%% Failed to init producer: %s\n", errstr);
    exit(1);
  }

  return 0;
}

初始化过程介绍:

  1. 首先通过rd_kafka_conf_new()函数创建rd_kafka_conf_t对象

  2. 设置rd_kafka_conf_t对象,设置kafka客户端参数,示例参数为:

    • bootstrap.servers: broker地址列表
    • security.protocol: 安全协议类型,示例为SASL_PLAINTEXT
    • sasl.mechanisms: sasl协议机制,示例为PLAIN, 表示普通文本
    • sasl.username: 认证用户名
    • sasl.password: 认证密码
    • api.version.request: 可选,librdkafka与kafka服务器版本适配参数,该参数为true表示允许librdkafka向broker发送请求询问broker支持的API版本列表(Apache Kafka v0.10.0版本后支持),以完成版本适配,更多版本适配要点见https://github.com/edenhill/librdkafka/wiki/Broker-version-compatibility
    • 设置发送消息的回调函数,因为librdkafka发送消息为非阻塞的,需要通过rd_kafka_poll()方法轮询消息是否发送成功,并设置响应的回调函数确认消息是否发送成功
  3. 调用rd_kafka_new()函数创建rd_kafka_t对象

发送消息

int KafkaApi::send_message(const std::string &topic, const char *data,
                           const int &data_len) {
  rd_kafka_topic_t *rkt = rd_kafka_topic_new(rk_, topic.c_str(), NULL);

  if (!rkt) {
    COMMLIB_LOG_ERR("kafka: create topic failed, err:%s",
                    rd_kafka_err2str(rd_kafka_errno2err(errno)));
    return rt::KDFKA_PRODUCE_ERR;
  }

  int ret = rd_kafka_produce(rkt, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY,
                             const_cast<char *>(data), data_len, NULL, 0, NULL);
  if (ret == -1) {
    COMMLIB_LOG_ERR("kafka: send message failed, err:%s",
                    rd_kafka_err2str(rd_kafka_errno2err(errno)));
    return rt::KDFKA_PRODUCE_ERR;
  }

  COMMLIB_LOG_DEBUG("produce message [%s]", data);
  rd_kafka_poll(rk_, 0);
  return rt::SUCCESS;
}

发送消息过程介绍:

  1. 通过rd_kafka_topic_new()方法创建rd_kafka_topic_t对象,注意topic是自动创建的(需要broker端设置能否自动创建topic的参数:auto.create.topics.enable=true), 除此之外,topic能否创建成功还与认证用户的权限有关,如果认证用户在broker端为super.users,则topic能够自动创建成功,否则则会报错: 用户无权限,需要先给用户添加ACL权限才行;最后一点,对于已经存在的topic, rd_kafka_topic_new()方法仍然返回的是旧的对象

  2. 发送消息通过调用rd_kafka_produce()函数完成,该函数的参数为:

    • rd_kafka_topic_t对象
    • partition: RD_KAFKA_PARTITION_UA表示为不设置
    • msgflags: 可设置为0或RD_KAFKA_MSG_F_COPY, RD_KAFKA_MSG_F_FREE, RD_KAFKA_MSG_F_BLOCK, RD_KAFKA_MSG_F_COPY表示发送的消息内容参数为值传递,rd_kafka_produce()函数返回之后将不会仍持有消息内容的引用
    • payload, 消息内容指针
    • len, 消息长度
    • key, 消息的key
    • msg_opaque: 每条消息的透明度指针,在消息发送的回调函数中使用
  3. 调用rd_kafka_poll()函数,使得消息发送的回调函数能够触发, 该函数第一个参数为rd_kafka_t对象,第二个参数为timeout_ms,设置为0表示为非阻塞

注意事项

在使用librdkafka带鉴权认证访问kafka服务器的过程中,解决消息发送失败问题的关键点有:

  • librdkafka的SASL依赖有没有添加
  • SASL认证的参数配置有没有正确,需要确认用户在broker端是否已经添加,以及确认用户拥有的权限
  • api.version.request参数,该参数设置不正确,将直接导致消息发送失败,使用过程中需要注意librdkafka的版本与broker的版本
    原文作者:bellengao
    原文地址: https://www.jianshu.com/p/3ede2c7ca250
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞