zoukankan      html  css  js  c++  java
  • 【转】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);

  • 相关阅读:
    SQL Server 2008 R2英文版安装图解
    浅析在C#里面抛出SAP里面自定义的异常信息
    JavaScript中的函数基础
    《冷眼看IT》读书笔记IT将成为服务行业
    JavaScript入门
    IT成为第五个服务业
    JavaScript中匿名函数的困惑
    自定义的html radio button的样式
    探索客户端JavaScript
    JavScript中的循环
  • 原文地址:https://www.cnblogs.com/swordenchanter/p/11404987.html
Copyright © 2011-2022 走看看