360开源又一力作——KafkaBridge：让操作kafka更简单！

待我称王封你为后i 2022-02-22 02:46 292阅读 0赞

## 导引 ##

\\\\

KafkaBridge 封装了对Kafka集群的读写操作，接口极少，简单易用，稳定可靠，支持c++/c、php、python、golang等多种语言，并特别针对php-fpm场景中作了长连接复用的优化，已在360公司内部广泛使用。

\\\\

## 前言 ##

\\\\

*  众所周知，Kafka是近几年来大数据领域最流行的分布式流处理平台。它最初由LinkedIn公司开发， 已于2010年贡献给了Apache基金会并成为顶级开源项目, 本质上是一种低延时的、可扩展的、设计内在就是分布式的，分区的和可复制的消息系统;\\\\t
 *  Kafka在360公司内部也有相当广泛的使用，业务覆盖搜索，商业广告，IOT, 视频，安全， 游戏等几乎所有核心业务，每天的写入流量近1.2PB, 读取流量近2.4PB;\\\\t
 *  Kafka官方提供了Java版本的客户端SDK, 但因360公司内部产品线众多，语言几乎囊括目前所有主流语言，所以我们研发了Kafka客户端SDK ——　KafkaBridge;\\

## 简介 ##

\\\\

*  KafkaBridge 底层基于 librdkafka, 与之相比封装了大量的使用细节，简单易用，使用者无需了解过多的Kafka系统细节，只需调用极少量的接口，就可完成消息的生产和消费;\\\\t
 *  针对使用者比较关心的消息生产的可靠性，作了近一步的提升；\\\\t
 *  开源地址：\[https://github.com/Qihoo360/kafkabridge\]\\

## 特点 ##

\\\\

*  支持多种语言：c++/c、php、python、golang, 且各语言接口完全统一;\\\\t
 *  接口少，简单易用;\\\\t
 *  针对高级用户，支持通过配置文件调整所有的librdkafka的配置;\\\\t
 *  在非按key写入数据的情况下，尽最大努力将消息成功写入;\\\\t
 *  支持同步和异步两种数据写入方式;\\\\t
 *  在消费时，除默认自动提交offset外，允许用户通过配置手动提交offset;\\\\t
 *  在php-fpm场景中，复用长连接生产消息，避免频繁创建断开连接的开销;\\

## 编译 ##

\\\\

*  编译依赖于 librdkafka, liblog4cplus, boost(仅依赖于若干个头文件);\\\\t
 *  对于C++/C使用 CMake 编译;\\\\t
 *  对于Python, Php, Golang使用 swig 编译;\\\\t
 *  每种语言都提供了自动编译脚本，方便使用者自行编译。\\

## 使用 ##

\\\\

#### 1、数据写入 ####

\\\\

*  在非按key写入的情况下，sdk尽最大努力提交每一条消息，只要Kafka集群存有一台broker正常，就会重试发送;\\\\t
 *  每次写入数据只需要调用*produce*接口，在异步发送的场景下，通过返回值可以判断发送队列是否填满，发送队列可通过配置文件调整;\\\\t
 *  在同步发送的场景中，*produce*接口返回当前消息是否写入成功，但是写入性能会有所下降，CPU使用率会有所上升,推荐还是使用异步写入方式;\\\\t
 *  我们来简单看一下写入kafka所涉及到的所有接口:\\

\//初始化接口\\bool QbusProducer::init(const string\u0026amp; broker_list, const string\u0026amp; log_path, const string\u0026amp; config_path, const string\u0026amp; topic)\\//写入数据接口\\bool QbusProducer::produce(const char* data, size_t data_len, const std::string\u0026amp; key)\\//不再需要写入数据时，需要调用的清理接口，必须调用\\void QbusProducer::uninit()

\\\\

*  具体使用可以参考源码中的实例;\\

#### 2、数据消费 ####

\\\\

*  消费只需调用subscribeOne订阅topic（也支持同时订阅多个topic），然后执行start就开始消费，当前进程非阻塞，每条消息通过callback接口回调给使用者;\\\\t
 *  sdk还支持用户手动提交offset方式，用户可以通过callback中返回的消息体，在代码其他逻辑中进行提交。\\\\t
 *  下面是消费接口，以c++为例：\\

\//初始化接口\\bool QbusConsumer::init(const string\u0026amp; string broker_list, const string\u0026amp; string log_path, const string\u0026amp; string config_path, QbusConsumerCallback\u0026amp; callback)\\//订阅需要消费的消息\\bool QbusConsumer::subscribeOne(const string\u0026amp; string group, const string\u0026amp; string topic)\\//开始消费\\bool QbusConsumer::start()\\//停止消费\\void QbusConsumer::stop()

\\\\

## 性能测试 ##

\\\\

*  kafka 集群三台broker, 除测试用topic外，无其他topic的读写操作;\\\\t
 *  测试用topic有3个partition;\\\\t
 *  Producer单实例，单线程;\\\\t
 *  Topic无复本下测试：\\\\t
    
    1.  单条消息 100 byte, 发送 1百万 条消息，耗时 1.7 秒;\\\\t\\t
    2.  单条消息 1024 byte, 发送 1百万 条消息，耗时 13 秒;\\\\t
    
    \\\\t
 *  Topic有2复本下测试：\\\\t
    
    1.  单条消息 100 byte, 发送 1百万 条消息，耗时 1.7 秒;\\\\t\\t
    2.  单条消息 1024 byte, 发送 1百万 条消息，耗时 14 秒;\\\\t
    
    \\

## 写在最后 ##

\\\\

*  KafkaBridge 一直在360公司内部使用，现在已经开源，有疏漏之处，欢迎广大使用者批评指正，也欢迎更多的使用者加入到 KafkaBridge 的持续改进中。\\\\t
 *  开源地址：KafkaBridge\\