Kafka开源项目指南

通过调用librdkafka库实现操作kafka

消费者：

#include <vector>
#include <string>
#include <memory>
#include <getopt.h>
#include <csignal>
#include <iostream>
#include "librdkafka/rdkafkacpp.h"
 
class kafka_consumer_client{
public:
    kafka_consumer_client(const std::string& brokers, const std::string& topics, std::string groupid, int64_t offset=-1);
    //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;
     
};

#include "kafka_comsumer.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";
    /*每次从单个分区中拉取消息的最大尺寸*/
    if(conf->set("max.partition.fetch.bytes", strfetch_num, errstr) != RdKafka::Conf::CONF_OK){
        fprintf(stderr, "RdKafka conf set max.partition failed : %s\n", errstr.c_str());
    }
 
    /*创建kafka consumer实例*/
    kafka_consumer_ = RdKafka::Consumer::create(conf, errstr);
    if(!kafka_consumer_){
        fprintf(stderr, "failed to ceate consumer\n");
    }
    delete conf;
 
    RdKafka::Conf *tconf = nullptr;
    /*创建kafka topic的配置*/
    tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
    if(!tconf){
        fprintf(stderr, "RdKafka create topic conf failed\n");
        return false;
    }
 
    /*kafka + zookeeper,当消息被消费时,会想zk提交当前groupId的consumer消费的offset信息,
    当consumer再次启动将会从此offset开始继续消费.在consumter端配置文件中(或者是
    ConsumerConfig类参数)有个"autooffset.reset"(在kafka 0.8版本中为auto.offset.reset),
    有2个合法的值"largest"/"smallest",默认为"largest",此配置参数表示当此groupId下的消费者,
    在ZK中没有offset值时(比如新的groupId,或者是zk数据被清空),consumer应该从哪个offset开始
    消费.largest表示接受接收最大的offset(即最新消息),smallest表示最小offset,即从topic的
    开始位置消费所有消息.*/
    if(tconf->set("auto.offset.reset", "smallest", errstr) != RdKafka::Conf::CONF_OK){
        fprintf(stderr, "RdKafka conf set auto.offset.reset failed : %s\n", errstr.c_str());
    }
 
    topic_ = RdKafka::Topic::create(kafka_consumer_, topics_, tconf, errstr);
    if(!topic_){
        fprintf(stderr, "RdKafka create topic failed : %s\n", errstr.c_str());
    }
    delete tconf;
 
    RdKafka::ErrorCode resp = kafka_consumer_->start(topic_, partition_, offset_);
    if (resp != RdKafka::ERR_NO_ERROR){
        fprintf(stderr, "failed to start consumer : %s\n", RdKafka::err2str(resp).c_str());
    }
 
    return true;
}
 
void kafka_consumer_client::consumer(RdKafka::Message *message, void *opt){
    switch(message->err()){
        case RdKafka::ERR__TIMED_OUT:
            break;
        case RdKafka::ERR_NO_ERROR:
            printf("%.*s\n", 
                static_cast<int>(message->len()),
             static_cast <const char*>(message->payload()));
            last_offset_ = message->offset();
          
             break;
        case RdKafka::ERR__PARTITION_EOF:
            std::cerr << "%% Reached the end of the queue, offset: " << last_offset_ << std::endl;
            break;
        case RdKafka::ERR__UNKNOWN_TOPIC:
        case RdKafka::ERR__UNKNOWN_PARTITION:
            std::cerr << "Consume failed: " << message->errstr() << std::endl;
            run_ = false;
            break;
        default:
            std::cerr << "Consume failed: " << message->errstr() << std::endl;
            run_ = false;
            break;
    }
}
 
bool kafka_consumer_client::consume(int timeout_ms){
    RdKafka::Message *msg = nullptr;
 
    while(run_){
        msg = kafka_consumer_->consume(topic_, partition_, timeout_ms);
        consumer(msg, nullptr);
        kafka_consumer_->poll(0);
        delete msg;
    }
 
    kafka_consumer_->stop(topic_, partition_);
    if(topic_){
        delete topic_;
        topic_ = nullptr;
    }
    if(kafka_consumer_){
        delete kafka_consumer_;
        kafka_consumer_ = nullptr;
    }
 
    /*销毁kafka实例*/
    RdKafka::wait_destroyed(5000);
    return true;
}
 

生产者：

#ifndef KAFKAPRODUCER_H
#define KAFKAPRODUCER_H
 
#include <iostream>
#include <string>
#include <cstdlib>
#include <cstdio>
#include <csignal>
#include <cstring>
#include <list>
#include <librdkafka/rdkafkacpp.h>
#include <vector>
#include <fstream>
 
using std::string;
using std::list;
using std::cout;
using std::endl;
using std::vector;
using std::fstream;
 
class KafkaProducerDeliveryReportCallBack : public RdKafka::DeliveryReportCb {
public:
    void dr_cb(RdKafka::Message &message) {
        std::cout << "Message delivery for (" << message.len() << " bytes): " <<
            message.errstr() << std::endl;
        if (message.key())
            std::cout << "Key: " << *(message.key()) << ";" << std::endl;
    }
};
class KafkaProducerEventCallBack : public RdKafka::EventCb {
public:
    void event_cb(RdKafka::Event &event) {
        switch (event.type())
        {
        case RdKafka::Event::EVENT_ERROR:
            std::cerr << "ERROR (" << RdKafka::err2str(event.err()) << "): " <<
                event.str() << std::endl;
            if (event.err() == RdKafka::ERR__ALL_BROKERS_DOWN)
            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;
        default:
            std::cerr << "EVENT " << event.type() <<
                " (" << RdKafka::err2str(event.err()) << "): " <<
                event.str() << std::endl;
            break;
        }
    }
};
class KafkaProducer
{
public:
    KafkaProducer(const string &brokers, const string &topics, int nPpartition = 0);
    virtual ~KafkaProducer();
    bool Init();
    void Send(const string &msg);
    void Stop();
private:
    RdKafka::Producer *m_pProducer = NULL;
    RdKafka::Topic *m_pTopic = NULL;
    KafkaProducerDeliveryReportCallBack m_producerDeliveryReportCallBack;
    KafkaProducerEventCallBack m_producerEventCallBack;
    std::string m_strTopics;
    std::string m_strBroker;
    bool m_bRun = false;
    int m_nPpartition = 0;
};
#endif // KAFKAPRODUCER_H

#include <iostream>
#include "kafkaproducer.h"
 
KafkaProducer::KafkaProducer(const string &brokers, const string &topics, int nPpartition /*= 1*/)
    : m_bRun(true), m_strTopics(topics), m_strBroker(brokers), m_nPpartition(nPpartition)
{
}
 
KafkaProducer::~KafkaProducer()
{
    Stop();
}
 
bool KafkaProducer::Init()
{
    string errstr = "";
 
    /*
     * Create configuration objects
     */
    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
    RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
 
    /*Set configuration properties,设置broker list*/
    if (conf->set("metadata.broker.list", m_strBroker, errstr) != RdKafka::Conf::CONF_OK){
        std::cerr << "RdKafka conf set brokerlist failed :" << errstr.c_str() << endl;
    }
    /* Set delivery report callback */
    conf->set("dr_cb", &m_producerDeliveryReportCallBack, errstr);
    conf->set("event_cb", &m_producerEventCallBack, errstr);
 
    /*
     * Create producer using accumulated global configuration.
    */
    m_pProducer = RdKafka::Producer::create(conf, errstr);
    if (!m_pProducer) {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        return false;
    }
    std::cout << "% Created producer " << m_pProducer->name() << std::endl;
    /*
     * Create topic handle.
    */
    m_pTopic = RdKafka::Topic::create(m_pProducer, m_strTopics,
                                      tconf, errstr);
    if (!m_pTopic) {
        std::cerr << "Failed to create topic: " << errstr << std::endl;
        return false;
    }
    return true;
}
void KafkaProducer::Send(const string &msg)
{
    if (!m_bRun)
        return;
    /*
     * Produce message
    */
    RdKafka::ErrorCode resp = m_pProducer->produce(m_pTopic, m_nPpartition,
                                                   RdKafka::Producer::RK_MSG_COPY /* Copy payload */,
                                                   const_cast<char *>(msg.c_str()), msg.size(),
                                                   NULL, NULL);
    if (resp != RdKafka::ERR_NO_ERROR)
        std::cerr << "Produce failed: " << RdKafka::err2str(resp) << std::endl;
    else
        std::cerr << "Produced message (" << msg.size() << " bytes)" << std::endl;
 
    m_pProducer->poll(0);
 
    /* Wait for messages to be delivered */  //firecat add
    while (m_bRun && m_pProducer->outq_len() > 0) {
        std::cerr << "Waiting for " << m_pProducer->outq_len() << std::endl;
        m_pProducer->poll(1000);
    }
}
 
void KafkaProducer::Stop()
{
    delete m_pTopic;
    delete m_pProducer;
}


 
int main()
{
    //KafkaProducerClient* KafkaprClient_ = new KafkaProducerClient("localhost:9092", "test", 0);
 
    KafkaProducer* Kafkapr_ = new KafkaProducer("localhost:9092", "test", 0);
    Kafkapr_->Init();
    Kafkapr_->Send("hello world!");
 
    char str_msg[] = "Hello Kafka!";
 
    while (fgets(str_msg, sizeof(str_msg), stdin))
    {
        size_t len = strlen(str_msg);
        if (str_msg[len - 1] == '\n')
        {
            str_msg[--len] = '\0';
        }
 
        if (strcmp(str_msg, "end") == 0)
        {
            break;
        }
 
        Kafkapr_->Send(str_msg);
    }
 
    return 0;
}