1.C++编程中使用librdkafka库去连接kafka集群分为生产端,消费端两个部分。本次是在https://github.com/edenhill/librdkafka 下载源码编译,安装的。过程很简单不再这里详细说明。
一.生产端使用
在编译完之后会有一个rdkafka_example.cpp,参考他进行编写程序。主要逻辑如下:
RdKafka::Conf *m_conf;
RdKafka::Conf *m_tconf;
RdKafka::Producer *m_producer;
RdKafka::Topic *m_topic;
MyHashPartitionerCb hash_partitioner;
ExampleDeliveryReportCb ex_dr_cb;
ExampleEventCb ex_event_cb;
m_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
m_tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
m_tconf->set("partitioner_cb", &hash_partitioner, errstr)
m_conf->set("metadata.broker.list", m_sBrokenList, errstr);
m_conf->set("queue.buffering.max.ms", "10", errstr); //批量提交
m_conf->set("queue.buffering.max.messages", sMaxMsgs, errstr);
m_conf->set("event_cb", &ex_event_cb, errstr);
m_conf->set("dr_cb", &ex_dr_cb, errstr);
m_producer = RdKafka::Producer::create(m_conf, errstr);
m_topic = RdKafka::Topic::create(m_producer, m_topicName,m_tconf, errstr);
RdKafka::ErrorCode resp = m_producer->produce(m_topic, m_PartionCn , RdKafka::Producer::RK_MSG_COPY, pData, iDataLen, (const void *)pKey, strlen(pKey),NULL);
m_producer->poll(1);//可以是poll 0
注意点:
1.成产过程中的pKey值需要注意一点,不能全部一样,这会影响到各分区的负载均衡。
2.queue.buffering.max.messages" //本地队列缓冲值,应该是在10w-50w之间,太小则会处理不过来引起丢弃。
3.在此基础上加上多线程就满足需求
二.消费端的使用
1.主要参考rdkafka_consumer_example.cpp 文件。
主要逻辑如下:
RdKafka::Conf *m_conf;
RdKafka::Conf *m_tconf;
m_conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
m_tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
ExampleRebalanceCb ex_rebalance_cb;
m_conf->set("rebalance_cb", &ex_rebalance_cb, errstr);
m_conf->set("enable.partition.eof", "true", errstr);
m_conf->set("metadata.broker.list", m_sBrokenList, errstr) //设置服务列表
m_conf->set("auto.commit.interval.ms", g_conf.sCommitTimeVal, errstr)
m_conf->set("group.id", m_sGroupID, errstr)
MyConsumeCb ex_consume_cb;
m_conf->set("consume_cb", &ex_consume_cb, errstr);
ExampleEventCb ex_event_cb;
m_conf->set("event_cb", &ex_event_cb, errstr);
m_tconf->set("auto.offset.reset", "latest", errstr);
m_conf->set("default_topic_conf", m_tconf, errstr);
RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(m_conf, errstr);
std::vectorstd::string topicsVec;
topicsVec.clear();
topicsVec.push_back(m_sTopicName);
RdKafka::ErrorCode err = consumer->subscribe(topicsVec);
while (m_runFlag)
{
RdKafka::Message *msg = consumer->consume(1000);//1000是超时时间单位毫秒
msg_consume(msg, NULL);
delete msg;
}
consumer->close();
delete consumer;
注意点:
1.消费者组中消费者数量不能大于分区数量。不同的消费者组可以消费相同topic
2.加上多线程即可满足多个消费者同时消费。
附上kafka参数的对照表