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的版本