设为首页 加入收藏

TOP

c++使用librdkafka库实现kafka的消费实例(一)
2017-08-10 10:22:42 】 浏览:3298
Tags:使用 librdkafka 实现 kafka 消费 实例

关于librdkafka库的介绍,可以参考kafka的c/c++高性能客户端librdkafka简介,本文使用librdkafka库来进行kafka的简单的消费

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);


完整代码

my_consumer.h:

#include 
  
   
#include 
   
     #include 
    
      #include 
     
       #include 
      
        #include 
       
         #include "../src-cpp/rdkafkacpp.h" class kafka_consumer_client{ public: kafka_consumer_client(const std::string& brokers, const std::string& topics, std::string groupid, int64_t offset); //kafka_consumer_client(); virtual ~kafka_consumer_client(); bool initClient(); bool consume(int timeout_ms); void finalize(); private: void consumer(RdKafka::Message *msg, void *opt); std::string brokers_; std::string topics_; std::string groupid_; int64_t last_offset_ = 0; RdKafka::Consumer *kafka_consumer_ = nullptr; RdKafka::Topic *topic_ = nullptr; int64_t offset_ = RdKafka::Topic::OFFSET_BEGINNING; int32_t partition_ = 0; };
       
      
     
    
   
  

my_consumer.cpp
#include "my_consumer_cpp.h"


bool run_ = true;

static void sigterm (int sig) {
  run_ = false;
}

kafka_consumer_client::kafka_consumer_client(const std::string& brokers, const std::string& topics, std::string groupid, int64_t offset)
:brokers_(brokers),
 topics_(topics),
 groupid_(groupid),
 offset_(offset){
 }

//kafka_consumer_client::kafka_consumer_client(){}

kafka_consumer_client::~kafka_consumer_client(){}

bool kafka_consumer_client::initClient(){
	RdKafka::Conf *conf = nullptr;
	conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
	if(!conf){
		fprintf(stderr, "RdKafka create global conf failed\n");
		return false;
	}

	std::string errstr;
	/*设置broker list*/
	if (conf->set("bootstrap.servers", brokers_, errstr) != RdKafka::Conf::CONF_OK){
		fprintf(stderr, "RdKafka conf set brokerlist failed : %s\n", errstr.c_str());
	}

	/*设置consumer group*/
	if (conf->set("group.id", groupid_, errstr) != RdKafka::Conf::CONF_OK){
		fprintf(stderr, "RdKafka conf set group.id failed : %s\n", errstr.c_str());
	}

	std::string strfetch_num = "10240000";
首页 上一页 1 2 3 下一页 尾页 1/3/3
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇C++ sha512 函数接口 下一篇C++的封装、继承、多态

最新文章

热门文章

Hot 文章

Python

C 语言

C++基础

大数据基础

linux编程基础

C/C++面试题目