使用librdkafka的C++接口实现Kafka生产者和消费者客户端

1. librdkafka简介：

librdkafka 是 Apache Kafka 的 C/C++ 开发包，提供 生产者、消费者 和 管理客户端。

设计理念是可靠以及高性能的消息传输，当前可支持每秒超过100万的消息生产和300万每秒的消息消费。

官方README 文档对librdkafka的介绍：
“librdkafka — the Apache Kafka c/C++ client library”

librdkafka/INTRODUCTION.md
https://github.com/edenhill/librdkafka/blob/master/INTRODUCTION.md

librdkafka/examples/
https://github.com/edenhill/librdkafka/tree/master/examples

Usage：
使用时，需要在源程序中包含包含 "rdkafka.h" 头文件

2. librdkafka的C++接口：

2.1 RdKafka::Conf::create()：

创建Conf配置实例，用于填充用户指定的各配置项：

//namespace RdKafka;

//brief Create configuration object:
//RdKafka::Conf ---> 配置接口类，用来设置对生产者、消费者、broker的各配置项的值
static Conf *create(ConfType type);	

enum ConfType {
	CONF_GLOBAL,	//Global configuration
	CONF_TOPIC		//Topic specific configuration
};

使用举例：

RdKafka::Conf *m_config = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
if(m_config == nullptr) {
	std::cout << "Create Rdkafka Global Conf Failed." << std::endl;
}

RdKafka::Conf *m_topicConfig = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
if(m_config == nullptr) {
	std::cout << "Create Rdkafka Topic Conf Failed." << std::endl;
}

2.2 Conf::ConfResult set()：

Conf类中的多个set成员函数，用于对不同的配置项进行赋值：

class Conf {
public:
	virtual Conf::ConfResult set(const std::string &name, const std::string &value, std::string &errstr);
	virtual Conf::ConfResult set(const std::string &name, DeliveryReportCb *dr_cb, std::string &errstr);
	virtual Conf::ConfResult set(const std::string &name, EventCb *event_cb, std::string &errstr);
	virtual Conf::ConfResult set(const std::string &name, PartitionerCb *partitioner_cb, std::string &errstr);
	//...
	//...
};

enum ConfResult {
	CONF_UNKNOWN = -2,	//Unknown configuration property
	CONF_INVALID = -1,	//Invalid configuration value
	CONF_OK = 0			//Configuration property was succesfully set
};

使用举例：

RdKafka::Conf::ConfResult  	result;
std::string					error_str;

RdKafka::Conf *m_config;
//设置 "booststrap::servers" 配置项：
result = m_config->set("bootstrap.servers", "127.0.0.`:9092", error_str);
if(result != RdKafka::Conf::CONF_OK) {
    std::cout << "Global Conf set 'booststrap.servers' failed: " << error_str << std::endl;
}

//设置 "event_cb" 配置项：
RdKafka::EventCb* m_event_cb = new ProducerEventCb;
result = m_config->set("event_cb", m_event_cb, error_str);
if(result != RdKafka::Conf::CONF_OK) {
    std::cout << "Global Conf set 'event_cb' failed: " << error_str << std::endl;
}

2.3 RdKafka::Producer::create()：

创建Producer生产者客户端：

class Producer : public virtual Handle {
public:
	static Producer *create(Conf *conf, std::string &errstr);
};

使用举例：

RdKfka::Producer *m_producer;

m_producer = RdKafka::Producer::create(m_config, error_str);
if(m_producer == nullptr) {
    std::cout << "Create Topic failed: " << error_str << std::endl;
}

2.4 RdKafka::Topic::create()：

创建Topic主题对象：

class Topic {
public:
	static Topic *create(Handle *base, const std::string &tipic_str, const Conf *conf, std::string &errstr);
};

使用举例：

RdKafka::Topic *m_topic;

m_topic = RdKafka::Topic::create(m_producer, m_topicStr, m_topicConfig, error_str);
if(m_topic == nullptr) {
    std::cout << "Create Topic failed: " << error_str << std::endl;
}

2.5 RdKafka::Producer::produce()：

class Producer : public virtual Handle {
public:
	virtual ErrorCode produce(Topic *topic, int32_t partition, int msgflags, 
						void *payload, size_t len, const std::string *key, void *msg_opaque);

	virtual ErrorCode produce();
};


//Use RdKafka::err2str() to translate an error code a human readable string
enum ErrorCode {
	//Internal errors to rdkafka:
	ERR_BEGIN = -200,		//Begin internal error codes
	ERR_BAD_MSG = -199,		//Received message is incorrect
	//...
	ERR_END = -100,			//End interval error codes

	//Kafka broker errors: 
	ERROR_UNKNOWN = -1,		//Unknown broker error
	ERROR_NO_ERROR = 0,		//Success
	//...
};

使用举例：

RdKafka::ErrorCode error_code = m_producer->produce(m_topic, RdKafka::Topic::PARTITION_UA, RdKafka::Producer::RK_MSG_COPY, 
										payload, len, key, NULL);

m_producer->pool(0);	//poll()参数为0意味着不阻塞，poll(0)主要是为了触发应用程序提供的回调函数

if(error_code != ERROR_NO_ERROR) {
	std::cerr << "Produce failed: " << RdKafka::err2str(error_code) << std::endl;
	if(error_code == ERR_QUEUE_FULL) {
		m_producer->poll(1000);		//如果发送失败的原始是队列正满，则阻塞等待一段时间
	}
	else if(error_code == ERR_MSG_SIZE_TOO_LARGE) {
		//如果消息过大超过了max_size，则需要对消息做裁剪后重新发送
	}
	else {
		std::cerr << "ERR_UNKNOWN_PARTITION or ERR_UNKNOWN_TOPIC" << std::endl;
	}
}

2.6 RdKafka::KafkaConsumer::create()：

创建Consumer消费者客户端：

class KafkaConsumer : public virtual Handle {
public:
	static KafkaConsumer *create(const Conf *conf, std::string &errstr);
};

使用举例：

RdKafka::KafkaConsumer *m_consumer;

m_consumer = RdKafka::KafkaConsumer::create(m_config, error_str);
if(m_consumer == nullptr) {
    std::cout << "Create KafkaConsumer failed: " << error_str << std::endl;
}

2.7 RdKafka::KafkaConsumer::subscribe()：

Consumer消费者订阅Topic主题：

class KafkaConsumer : public virtual Handle {
public:
	virtual ErrorCode subscribe(const std::vector<std::string> &topics);
};

使用举例：

std::vector<std::string> topics;
topics.push_back(topic_str);

RdKafka::ErrorCode error_code = m_consumer->subscribe(topics);
if(error_code != ERROR_NO_ERROR) {
	std::cerr << "Consumer Subscribe Topics Failed: " << RdKafka::err2str(error_code) << std::endl;
}

2.8 RdKafka::KafkaConsumer::consume()：

Consumer消费者拉取消息进行消费：

class KafkaConsumer : public virtual Handle {
public
	virtual Message *consume(int timeout_ms);
};

使用举例：

RdKafka::Message *m_message = m_consumer->consume(5000);	//若超过 5000ms 未订阅到消息，则触发 RdKafka::ERR_TIMED_OUT

2. 使用librdkafka的C++接口实现生产者客户端：

2.1 main_producer.cpp

#include "producer_kafka.h"

using namespace std;


int main() {
    KafkaProducer producer("127.0.0.1:9092", "topic-demo", 0);

    sleep(5);

    for(int i = 0; i < 10; i++) {
        char msg[64] = {0};
        sprintf(msg, "%s%4d", "Hello Kafka ", i);   //msg = "Hello Kafka 0001";

        char key[8] = {0};
        sprintf(key, "%d", i);  //key = "1";

        producer.pushMessage(msg, key);
    }

    KafkaProducer::wait_destroyed(50000);

    return 0;
}

2.2 kafka_producer.h

#ifndef __KAFKAPRODUCER_H_
#define __KAFKAPRODUCER_H_

#include <string>
#include <iostream>
#include "rdkafkacpp.h"


class KafkaProducer {
public:
    explicit KafkaProducer(const std::string& brokers, const std::string& topic, int partition);       //epplicit：禁止隐式转换，例如不能通过string的构造函数转换出一个broker
    ~KafkaProducer();                                                                                                       

    void pushMessage(const std::string& msg, const std::string& key);



protected:
    std::string     m_brokers;
    std::string     m_topicStr;
    int             m_partition;

    RdKafka::Conf*      m_config;           //RdKafka::Conf --- 配置接口类，用来设置对 生产者、消费者、broker的各项配置值
    RdKafka::Conf*      m_topicConfig;

    RdKafka::Producer*  m_producer;
    RdKafka::Topic*     m_topic;
    
    RdKafka::DeliveryReportCb*      m_dr_cb;            //RdKafka::DeliveryReportCb 用于在调用 RdKafka::Producer::produce() 后返回发送结果，RdKafka::DeliveryReportCb是一个类，需要自行填充其中的回调函数及处理返回结果的方式
    RdKafka::EventCb*               m_event_cb;         //RdKafka::EventCb 用于从librdkafka向应用程序传递errors,statistics,logs 等信息的通用接口
    RdKafka::PartitionCb*           m_partitioner_cb;   //Rdkafka::PartitionCb 用于设定自定义分区器


};




class ProducerDeliveryReportCb : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message &message) {  //重载基类RdKafka::DeliveryReportCb中的虚函数dr_cb()
        if(message.err() != 0) {       //发送出错
            std::cerr << "Message delivery failed: " << message.errstr() << std::endl;
        } else {                             //发送成功
            std::cerr << "Message delivered to topic: " << message.topic_name() 
                      << " [" << message.partition() 
                      << "] at offset " << message.offset() << std::endl;
        }
    }
};
/*
class DeliveryReportCb {                        //Kafka将会在produce()之后返回发送结果时调用 DeliveryReportCb::dr_cb(), 并将结果填充到message中
public:
    virtual void dr_cb(Message &message) = 0;   //Message::err() 中保存返回结果
    virtual ~DeliveryReportCb() {}
};

返回的发送结果 Message 中包含的信息：
class Message {
public:
    virtual std::string     errstr() const = 0;     //返回错误原因
    virtual ErrorCode       err() const = 0;        //返回错误码，ErrorCode是enum枚举类型
    virtual Topic           *topic() const = 0;     //返回生产的消息所发送到的主题，返回值类型为 RdKafka::Topic 类对象
    virtual std::string     topic_name() const = 0; //返回生产的消息所发送到的主题名
    virtual int32_t         partition() const = 0;  //返回生产的消息所发送到的分区号

    virtual void            *payload() const = 0;   //消息内容
    virtual size_t          len() const = 0;        //消息长度
    virtual const std::string *key() const = 0;     //消息的key
    virtual int64_t         offset() const = 0;     //消息偏移量
    //...
};

enum ErrorCode {
    ERR_BEGIN = -200,
    //...
    ERR_UNKNOWN = -1,
    ERR_NO_ERROR = 0,   //SUCCESS
    ERR_OFFSET_OUT_OF_RANGE = 1,
    //...
};
*/


class ProducerEventCb : public RdKafka::EventCb {
public:
    void event_cb(RdKafka::Event &event) {
        switch(event.type()) {
            case RdKafka::EVENT::EVENT_ERROR:
                std::cout << "RdKafka::EVENT::EVENT_ERROR: " << RdKafka::err2str(event.err()) << std::endl;
                break;
            case RdKafka::EVENT::EVENT_STATS:
                std::cout << "RdKafka::EVENT::EVENT_STATS: " << events.str() << std::endl;
                break;
            case RdKafka::EVENT::EVENT_LOG:
                std::cout << "RdKafka::EVENT::EVENT_LOG: " << events.fac() << std::endl;
                break;
            case RdKafka::EVENT::EVENT_THROTTLE:
                std::cout << "RdKafka::EVENT::EVENT_THROTTLE: " << event.broker_name() << std::endl;
                break;
        }

    }
}
/*
EventCb 是用于从librdkafka向应用程序返回errors,statistics, logs等信息的通用接口：

class EventCb {
public:
    virtual void event_cb(Event &event) = 0;        //提供给应用程序去自定义重载的event_cb函数
    virtual ~EventCb(){ }
};

class Event {
public:
    enum Type {
        EVENT_ERROR,
        EVNET_STATS,
        EVENT_LOG,
        EVENT_THROTTLE      //Event is a throttle level signaling from the broker, 紧急事件?
    };

    enum Severity {         //表示Event事件的严重级别
        EVENT_SEVERITY_EMERGE = 0,
        EVENT_SEVERITY_ALERT = 1,
        EVENT_SEVERITY_CRITICAL = 2,
        EVENT_SEVERITY_ERROR = 3,
        EVENT_SEVERITY_WARNING = 4,
        EVENT_SEVERITY_NOTICE = 5,
        EVENT_SEVERITY_INFO = 6,
        EVENT_SEVERITY_DEBUG = 7,
    };

    virtual Type            type() const = 0;       //返回Event事件类型
    virtual ErrorCode       err() const = 0;        //错误码
    virtual Severity        severity() const = 0;   //return log serverity level, 返回log的严重级别
    virtual std::string     fac() const = 0;

    virtual std::string     broker_name() const = 0;
    virtual std::string     broker_id() const = 0;

	virtual vool 			fatal() const = 0;		//bool值，返回这是否是一个fatal级错误
    //...
};
*/


class HashPartitionerCb : public RdKafka::PartitionerCb {       //自定义生产者分区器，作用就是返回一个分区id。  对key计算Hash值，得到待发送的分区号（其实这跟默认的分区器计算方式是一样的）
public:
    int32_t partitioner_cb( const Topic *topic, const std::string *key, 
                            int32_t partition_cnt, void *msg_opaque)    
    {
        char msg[128] = {0};
        sprintf(smg, "HashPartitionCb:[%s][%s][%d]", topic->name().c_str(), key->c_str(), partition_cnt);
        std::cout << msg << std::endl; 

        //前面的操作只是为了在分区器回调中打印出一行打印，分区器真正的操作是在下面generate_hash，生成一个待发送的分区ID
        return generate_hash(key->c_str(), key->size()) % partition_cnt;    
    }
private:
    static inline unsigned int generate_hash(const char *str, size_t len) {
        unsigned int hash = 5381;
        for (size_t i = 0; i < len; i++) {
            hash = ( (hash << 5) + hash ) + str[i];
        }
        return hash;    //返回值必须在 0 到 partition_cnt 之间。如果出错则发回 PARTITION_UA（-1）
    }
};
/*
class PartitionerCb {
public:
    virtual int32_t partitioner_cb( const Topic *topic, 
                                    const std::string *key, 
                                    int32_t partition_cnt, 
                                    void *msg_opaque) = 0;      //自定义分区器的接口，需要指定发送的主题、key、分区数、opaque
    virtual ~PartitionerCb() { }
};

注意：
用户实现的派生类重载partitioner_cb() 这个函数后，也是要提供给Kafka去调用的，其中参数 partition_cnt 并非由Producer指定，而是Kafka根据Topic创建时的信息去查询，
且Kafka上的Topic创建也不是由Producer生产者客户端创建的，目前已知的方法只有使用 kafka-topics.sh 脚本这一种方法。

关于“创建主题”的描述：
-- 如果broker端配置参数 auto.create.topics.enable 设置为true（默认值为true），
那 么当生产者向一个尚未创建的主题发送消息时，会自动创建一个分区数为 num.partitions（默认值为1）、副本因为为 default.replication.factor（默认值为1）的主题。
-- 除此之外，当一个消费者开始从未知主题中读取消息时，或者当任意一个客户端向未知主题发送元数据请求时，都会按照配置参数 num.partitions 和 default.replication.factor的值创建一个相应主题。
*/

#endif


/*
总结：
1. KafkaProducer构造函数：初始化一个“生产者客户端”，要指定三个参数：（1）生产者所要连往的Kafka地址，是zookeeper的ip和port；（2）生产者后续生产的消息所要发送的Topic主题；（3）消息所要发往的分区？

2. RdKafka::Conf 是配置接口类，用来设置生产者、消费者、broker的各项配置值：
namespace RdKafka {
    RdKafka::Conf {
        enum ConfType {
            CONF_GLOBAL,    //Global Configuration , 全局级别的配置 （broker级，所有主题都一样）
            CONF_TOPIC      //Topic specfic configuration , Topic主题级别的专用配置（每个主题都不一样）
        };

        enum ConfResult {           //用于保存 RdKafka::Conf::set() 函数的返回值
            CONF_UNKNOWN = -2,      //Unknown configuration property, set() 传入的配置项不合法（Kafka中没有定义对应的配置项）
            CONF_INVALID = -1,      //Invalid configuration value, set() 传入的配置项的值不正确
            CONF_OK = 0             //Configuration property was successfully set, set() 设置成功
        };

        static Conf *create(ConfType type);     //创建一个Conf实例，传入ConfType参数，返回Conf*实例的指针，是static静态函数

        virtual Conf::ConfResult set(const std::string &name, const std::string &value, std::string &errstr);   //配置name指定的配置项的值
                //eg: RdKafka::Conf::ConfResult errCode = m_config->set("dr_cb", m_dr_cb, errorstr);

        
    };
};



Q: 为什么要创建 RdKafka::Conf(GLOBAL/TOPIC)、RdKafka::Producer、RdKafka::Topic ？
A: 这就跟socket() 函数一样，创建一个sockfd句柄，是为了通知内核创建相关的实例，sockfd是与内核中的实例相关联的句柄，内核需要维护不同的实例来区分不同的连接。
同理，Kafka服务器也需要区分不同的生产者，需要区分不同的生产者的差异化配置。因此需要应用程序创建一个Producer客户端实例（m_producer句柄），将自己想要的配置填充好（m_conf），
再将m_producer句柄和m_conf配置属性都传递给Kafka服务端，Kafka在内部创建并维护这个生产者实例，当生产者往Kafka的broker上推送消息时，Kafka就可以根据预先配置好的属性对其进行相应的处理。

*/

2.3 producer_kafka.cpp

#include "producer_kafka.h"


//("192.168.0.105:9092", "topic_demo", 0)
KafkaProducer::KafkaProducer(const std::string& brokers, const std::string& topic, int partition) {
    m_brokers   = brokers;
    m_topicStr  = topic;
    m_partition = partition;

    //先填充构造生产者客户端的参数配置：
    m_config = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    if(m_config == nullptr) {
        std::cout << "Create Rdkafka Global Conf Failed." << std::endl;
    }

    m_topicConfig = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
    if(m_topicConfig == nullptr) {
        std::cout << "Create Rdkafka Topic Conf Failed." << std::endl;
    }

    //下面开始配置各种需要的配置项：
    RdKafka::Conf::ConfResult   result;
    std::string                 error_str;
    
    result = m_config->set("booststrap.servers", m_brokers, error_str); //设置生产者待发送服务器的地址: "ip:port" 格式
    if(result != RdKafka::Conf::CONF_OK) {
        std::cout << "Global Conf set 'booststrap.servers' failed: " << error_str << std::endl;
    }

    result = m_config->set("statistics.interval.ms", "10000", error_str);
    if(result != RdKafka::Conf::CONF_OK) {
        std::cout << "Global Conf set ‘statistics.interval.ms’ failed: " << error_str << std::endl;
    }

    result = m_config->set("message.max.bytes", "10240000", error_str);     //设置发送端发送的最大字节数，如果发送的消息过大则返回失败
    if(result != RdKafka::Conf::CONF_OK) {
        std::cout << "Global Conf set 'message.max.bytes' failed: " << error_str << std::endl;
    }


    m_dr_cb = new ProducerDeliveryReportCb;
    result = m_config->set("dr_cb", m_dr_cb, error_str);    //设置每个消息发送后的发送结果回调
    if(result != RdKafka::Conf::CONF_OK) {
        std::cout << "Global Conf set ‘dr_cb’ failed: " << error_str << std::endl;
    }

    m_event_cb = new ProducerEventCb;
    result = m_config->set("event_cb", m_event_cb, error_str);
    if(result != RdKafka::Conf::CONF_OK) {
        std::cout << "Global Conf set ‘event_cb’ failed: " << error_str << std::endl;
    }

    m_partitioner_cb = new HashPartitionerCb;
    result = m_topicConfig->set("partitioner_cb", m_partitioner_cb, error_str);     //设置自定义分区器
    if(result != RdKafka::Conf::CONF_OK) {
        std::cout << "Topic Conf set ‘partitioner_cb’ failed: " << error_str << std::endl;
    }

    //创建Producer生产者客户端：
    m_producer = RdKafka::Producer::create(m_config, error_str);    //RdKafka::Producer::create(const RdKafka::Conf *conf, std::string &errstr);
    if(m_producer == nullptr) {
        std::cout << "Create Producer failed: " << error_str << std::endl;
    }

    //创建Topic对象，后续produce发送消息时需要使用
    m_topic = RdKafka::Topic::create(m_producer, m_topicStr, m_topicConfig, error_str); //RdKafka::Topic::create(Hanle *base, const std::string &topic_str, const Conf *conf, std::string &errstr);
    if(m_topic == nullptr) {
        std::cout << "Create Topic failed: " << error_str << std::endl;
    }
}

/*
另外几个关键的参数：
partition.assignment.strategy   : range,roundrobin  消费者客户端partition分配策略，当被选举为leader时，分配partition分区给组员消费者的策略
*/


void KafkaProducer::pushMessage(const std::string& msg, const std::string& key) {
    int32_t len = str.length();
    void *payload = const_cast<void*>(static_cast<const void*>(str.data()));

    RdKafka::ErrorCode error_code = m_producer->prodce( m_topic, RdKafka::Topic::PARTITION_UA,
                                                        RdKafka::Producer::RK_MSG_COPY,
                                                        payload, len, key, NULL);
    m_producer->poll(0);        //poll()参数为0意味着不阻塞；poll(0)主要是为了触发应用程序提供的回调函数
    if(error_code != ERR_NO_ERROR) {
        std::cerr << "Produce failed: " << RdKafka::err2str(error_code) << std::endl;
        if(error_code == ERR_QUEUE_FULL) {
            m_producer->poll(1000);     //如果发送失败的原因是队列正满，则阻塞等待一段时间
        } else if(error_code == ERR_MSG_SIZE_TOO_LARGE) {
            //如果发送消息过大，超过了max.size，则需要裁减后重新发送
        } else {
            std::cerr << "ERR_UNKNOWN_PARTITION or ERR_UNKNOWN_TOPIC" << std::endl;
        }
    }
}
/*
RdKafka::ErrorCode RdKafka::Produce::produce(   RdKafka::Topic *topic, int32_t partition,   //生产消息发往的主题、分区
                                                int msgflags,                               //RK_MSG_COPY（拷贝payload）、 RK_MSG_FREE（发送后释放payload内容）、 RK_MSG_BLOCK（在消息队列满时阻塞produce函数）
                                                void *payload, size_t len,                  //消息净荷、长度
                                                const string &key, void *msg_opaque);       //key; opaque是可选的应用程序提供给每条消息的opaque指针，opaque指针会在dr_cb回调函数内提供，在Kafka内部维护
返回值：
    ERR_NO_ERROR            :   消息成功入队
    ERR_QUEUE_FULL          :   队列满
    ERR_MSG_SIZE_TOO_LARGE  :   消息长度过大
    ERR_UNKNOWN_PARTITION   :   所指定的分区不存在
    ERR_UNKNOWN_TOPIC       :   所指定的主题不存在


int RdKafka::Producer::poll (int timeout_ms);
阻塞等待生产消息发送完成，
poll另外的重要作用是：
（1）轮询处理指定的Kafka句柄的Event（m_producer上的Event事件，在本例中处理事件的方式是进行打印）；
（2）触发应用程序提供的回调函数调用，例如 ProducerDeliveryReportCb 等回调函数都需要poll()进行触发。
*/


KafkaProducer::~KafkaProducer() {
    while(m_producer->outq_len() > 0) {         //当 Handle->outq_len() 客户端的“出队列” 的长度大于0
        std::cerr << "Waiting for: " << m_producer->outq_len() << std::endl;
        m_producer->flush(5000);
    }

    delete m_config;
    delete m_topicConfig;
    delete m_topic;
    delete m_producer;
    delete m_dr_cb;
    delete m_event_cb;
    delete m_partitioner_cb;
}

/*
-- flush():
ErrorCode Kafka::Producer::flush(int timeout_ms);
flush会优先调用poll()，去触发生产者提前注册的各种回调函数，然后等待生产者上的所有消息全部发送完毕。


-- outq_len: “出队列”长度，是 Handle 类的一个成员，
表示 生产者队列中中待发送到broker上的数据，或 消费者队列中待发送到broker上的ACK。

Handle是Producer和Consumer类的基类，表示“客户端的句柄”：
    class Producer : public virtual Handle { }
    class Consumer : public virtual Handle { }
    class KafkaConsumer : public virtual Handle { }
*/

3. 使用librdkafka的C++接口实现实现消费者客户端：

3.1 main_consumer.cpp

#include "kafka_consumer.h"

int main()
{
    std::string brokers = "127.0.0.1:9092";
    
    std::vector<std::string> topics;		//待消费主题的集合
    topics.push_back("topic-demo");

    std::string group = "consumer-group-demo";	//消费组
    
    KafkaConsumer consumer(brokers, group, topics, RdKafka::Topic::OFFSET_BEGINNING);
    
    consumer.pullMessage();

    RdKafka::wait_destroyed(5000);
    
    return 0;
}

/*
在生产者/消费者 客户端 连接 broker 时，填充“bootstrap.server” 参数：
	Q: 为什么只设置了一个broker的地址（port = 9092），如果Kafka集群中有多个broker，且生产者/消费者订阅的Topic横跨多个broker时，生产者是如何知道其他broker的？
	A: bootstrap.server 参数用来指定生产者客户端连接Kafka集群所需的broker地址清单，具体的内容格式为：
	host1:port1, host2:port2
	可以设置一个或多个地址，中间以逗号隔开，此参数的默认值为 " "。
	【注意这里并非需要所有的broker地址，因为生产者会从给定的broker里查找到其他broker信息。】
	不过建议至少要设置两个以上的broker地址信息，当其中任意一个宕机时，生产者仍然可以连接到Kafka集群上。
*/

3.2 kafka_consumer.h

#ifndef __KAFKACONSUMER_H_
#define __KAFKACONSUMER_H_

#include <string>
#include <iostream>
#include <vector>
#include <stdio.h>
#include "rdkafkacpp.h"

class KafkaConsumer {
public:
    explicit KafkaConsumer(const std::string& brokers, const std::string& groupID,
                           const std::vector<std::string>& topics, int partition);
    ~KafkaConsumer();
    
    void pullMessage();
    
protected:
    std::string 				m_brokers;
    std::string 				m_groupId;
    std::vector<std::string> 	m_topicVector;		//一个消费者可以同时订阅多个主题，所有用vector
    int 						m_partition;
    
    RdKafka::Conf* 				m_config;			//GLOBAL 级别的配置（Consumer客户端级别）
    RdKafka::Conf* 				m_topicConfig;		//TOPIC	级别的配置
    
    RdKafka::KafkaConsumer* 	m_consumer;			//消费者客户端实例
    
    RdKafka::EventCb* 			m_event_cb;			//Event事件回调
    RdKafka::RebalanceCb* 		m_rebalance_cb;		//再均衡 回调
};


class ConsumerEventCb : public RdKafka::EventCb {
public:
    void event_cb(RdKafka::Event& event) {
        switch (event.type())
        {
	        case RdKafka::Event::EVENT_ERROR:
	            if (event.fatal()) 				//判断是否为FATAL错误
	                std::cerr << "FATAL ";
	            std::cerr << "ERROR (" << RdKafka::err2str(event.err()) << "): " << event.str() << std::endl;
	            break;
	        case RdKafka::Event::EVENT_STATS:
	            std::cerr << "\"STATS\": " << event.str() << std::endl;
	            break;
	        case RdKafka::Event::EVENT_LOG:
	            fprintf(stderr, "LOG-%i-%s: %s\n", event.severity(), event.fac().c_str(), event.str().c_str());
	            break;
	        case RdKafka::Event::EVENT_THROTTLE:
	            std::cerr << "THROTTLED: " << event.throttle_time() << "ms by " << event.broker_name() << " id " << (int)event.broker_id() << std::endl;
	            break;
	        default:
	            std::cerr << "EVENT " << event.type() << " (" << RdKafka::err2str(event.err()) << "): " <<  event.str() << std::endl;
	            break;
        }
    }
};

class ConsumerRebalanceCb : public RdKafka::RebalanceCb {
public:
    void rebalance_cb( RdKafka::KafkaConsumer *consumer, RdKafka::ErrorCode err,
                       std::vector<RdKafka::TopicPartition*> &partitions)		//Kafka服务端通过 err参数传入再均衡的具体事件（发生前、发生后），通过partitions参数传入再均衡 前/后，旧的/新的 分区信息
    {
        std::cerr << "RebalanceCb: " << RdKafka::err2str(err) << ": " << printTopicPartition(partitions);
        
        if(err == RdKafka::ERR__ASSIGN_PARTITIONS) {		//ERR__ASSIGN_PARTITIONS: 表示“再均衡发生之后，消费者开始消费之前”，此时消费者客户端可以从broker上重新加载offset
            consumer->assign(partitions);					//再均衡后，重新 assign() 订阅这些分区
            partition_count = (int)partitions.size();
        } else if(err == RdKafka::ERR__REVOKE_PARTITIONS) {		//ERR__REVOKE_PARTITIONS: 表示“消费者停止消费之后，再均衡发生之前”，此时应用程序可以在这里提交 offset
            consumer->unassign();								//再均衡前，unassign() 退订这些分区
            partition_count = 0;								//退订所有分区后，清0
        } else {
			std::cerr << "Rebalancing error: " << RdKafka::err2str(err) << std::endl;
		}
    }
    
private:
    static void printTopicPartition(const std::vector<RdKafka::TopicPartition*> &partitions) {	//打印出所有的主题、分区信息
        for(unsigned int i = 0 ; i < partitions.size() ; i++) {
            std::cerr << partitions[i]->topic() << "[" << partitions[i]->partition() << "], ";
        }
        std::cerr << "\n";
    }
private:
    int partition_count;			//保存consumer消费者客户端 当前订阅的分区数
};

/*
class RdKafka::RebalanceCb {
public:
	virtual void rebalance_cb(	RdKafka::KafkaConsumer *consumer, RdKafka::ErrorCode err, 
								std::vector<TopicPartition*> &partitions);	
	virtual ~RebalanceCb() { }
};

注意参数 vector<TopicPartition*> &partitions; 中 元素的类型是 TopicPartiton：
	class TopicPartitionImpl {
		string 	topic;
		int 	partition_;
	};
同时包括 主题 和 分区信息，所以 consumer.assign(); 订阅分区的方式是包括不同主题的不同分区的集合。
*/

#endif

3.3 kafka_consumer.cpp

#include "kafka_consumer.h"

KafkaConsumer::KafkaConsumer(const std::string& brokers, const std::string& groupId, 
							 const std::vector<std::string>& topics, int partition) 
{
	m_brokers		= borker;
	m_groupId		= groupId;
	m_topicVector	= topics;
	m_partition		= partition;

	//创建Conf实例：
	m_config = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
	if(m_config == nullptr) {
		std::cout << "Create Rdkafka Global Conf Failed." << std::endl;
	}

	m_topicConfig = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
	if(m_topicConfig == nullptr) {
		std::cout << "Create Rdkafka Topic Conf Failed." << std::endl;
	}

	//设置Conf的各个配置参数：
	RdKafka::Conf::ConfResult   result;
    std::string                 error_str;

    result = m_config->set("bootstrap.servers", m_brokers, error_str);
    if(result != RdKafka::Conf::CONF_OK) {
    	std::cout << "Conf set 'bootstrap.servers' failed: " << error_str << std::endl;
    }

    result = m_config->set("group.id", m_groupId, error_str);		//设置消费组名：group.id（string类型）
    if(result != RdKafka::Conf::CONF_OK) {
    	std::cout << "Conf set 'group.id' failed: " << error_str << std::endl;
    }

    result = m_config->set("max.partition.fetch.bytes", "1024000", error_str);	//消费消息的最大大小
    if(result != RdKafka::Conf::CONF_OK) {
    	std::cout << "Conf set 'max.partition.fetch.bytes' failed: " << error_str << std::endl;
    }

    result = m_config->set("enable.partition.eof", "false", error_str);		//enable.partition.eof: 当消费者到达分区结尾，发送 RD_KAFKA_RESP_ERR__PARTITION_EOF 事件，默认值 true
    if(result != RdKafka::Conf::CONF_OK) {
    	std::cout << "Conf set 'enable.partition.eof' failed: " << error_str << std::endl;
    }


    m_event_cb = new ConsumerEventCb;
    result = m_config->set("event_cb", m_event_cb, error_str);
    if(result != RdKafka::Conf::CONF_OK) {
    	std::cout << "Conf set 'event_cb' failed: " << error_str << std::endl;
    }

    m_reblance_cb = new ConsumerRebalanceCb;
    result = m_config->set("rebalance_cb", m_reblance_cb, error_str);
    if(result != RdKafka::Conf::CONF_OK) {
    	std::cout << "Conf set 'rebalance_cb' failed: " << error_str << std::endl;
    }


    //设置 topic_conf的配置项：
    result = m_topicConfig->set("auto.offset.reset", "latest", error_str);
    if(result != RdKafka::Conf::CONF_OK)
    {
        std::cout << "Topic Conf set 'auto.offset.reset' failed: " << error_str << std::endl;
    }

    result = m_config->set("default_topic_conf", m_topicConfig, error_str);
    if(result != RdKafka::Conf::CONF_OK)
    {
        std::cout << "Conf set 'default_topic_conf' failed: " << error_str << std::endl;
    }


    //创建消费者客户端：
    m_consumer = RdKafka::KafkaConsumer::create(m_config, error_str);
    if(m_consumer == nullptr) {
    	std::cout << "Create KafkaConsumer failed: " << error_str << std::endl;
    }
    std::cout << "Create KafkaConsumer succeed, consumer name : " << m_consumer->name() << std::endl;
}


void RdKafkaConsumer::pullMessage() {

}