【转】c++(11)使用librdkafka库实现kafka的消费实例

版权声明:本文为博主原创文章,遵循 CC 4.0 by-sa 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/lijinqi1987/article/details/76691170

librdkafka在c语言的基础上封装了一层c++的API,可以实现kafka的消费操作,基本操作步骤如下


1、创建kafka 配置

RdKafka::Conf *conf = nullptr;
conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

2、设置kafka各项参数


/*设置broker list*/
conf->set("bootstrap.servers", brokers_, errstr);

/*设置consumer group*/
conf->set("group.id", groupid_, errstr);

/*每次从单个分区中拉取消息的最大尺寸*/
conf->set("max.partition.fetch.bytes", strfetch_num, errstr);


3、创建kafka topic配置
RdKafka::Conf *tconf = nullptr;
tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);


4、设置kafka topic参数

if(tconf->set("auto.offset.reset", "smallest", errstr)


5、创建kafka consumer实例

kafka_consumer_ = RdKafka::Consumer::create(conf, errstr);


6、创建kafka topic

RdKafka::Topic::create(kafka_consumer_, topics_, tconf, errstr);


7、启动kafka consumer实例

RdKafka::ErrorCode resp = kafka_consumer_->start(topic_, partition_, offset_);


8、消费kafka

kafka_consumer_->consume(topic_, partition_, timeout_ms);


9、阻塞等待消息

kafka_consumer_->poll(0);


10、停止消费

kafka_consumer_->stop(topic_, partition_);


11、销毁consumer实例

RdKafka::wait_destroyed(5000);

原文地址:https://www.cnblogs.com/swordenchanter/p/11404987.html