devbox
酷酷是懒虫
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

kafka基本原理及C/C++ API 实现_kafka c接口

作者：酷酷是懒虫 | 2024-07-28 11:45:25

赞

踩

kafka c接口

引入消息队列的作用：

  1. 解耦，不同服务组件通过消息队列实现接口
  2. 缓冲，利于消费者和生产者速度匹配，也利于流量削峰
  3. 异步，消息队列可以实现将

Partition：为了实现扩展性，提⾼并发能⼒，⼀个⾮常⼤的 Topic 可以分布到多个 Broker（即服务器）上，⼀个 Topic 可以分为多个 Partition，同⼀个topic在不同的分区的数据是不重复的，每个 Partition 是⼀个有序的队列，其表现形式就是⼀个⼀个的⽂件夹。
 

Replication：每⼀个分区都有多个副本，副本的作⽤是做备胎。当主分区（Leader）故障的时候会选择⼀个备胎（Follower）上位，成为Leader。在kafka中默认副本的最⼤数量是10个，且副本的数量不能⼤于Broker的数量，follower和leader绝对是在不同的机器，同⼀机器对同⼀个分区也只可能存放⼀个副本（包括⾃⼰）

Leader：每个分区多个副本的“主”副本，⽣产者发送数据的对象，以及消费者消费数据的对象，都是 Leader。


Follower：每个分区多个副本的“从”副本，实时从 Leader 中同步数据，保持和 Leader 数据的同步。Leader 发⽣故障时，某个 Follower 还会成为新的 Leader。

以下，展示kafka中，一个主题TopicA中，三个patition在不同的broker中的分布，以及生产者和消费者是如何分配的。

C/C++接口实现

libkafka C接口实现：

kafkaConsumer.h

  1. #ifndef KAFKACONSUMER_H
  2. #define KAFKACONSUMER_H
  3.  
  4. #include <string>
  5. #include <iostream>
  6. #include <vector>
  7. #include <stdio.h>
  8.  
  9. #include "rdkafka.h"
  10.  
  11. using namespace std;
  12.  
  13. class KafkaConsumer
  14. {
  15. public:/**
  16.      * @brief KafkaConsumer
  17.      * @param brokers
  18.      * @param groupID
  19.      * @param topics
  20.      * @param partition
  21.      */
  22.     KafkaConsumer();
  23.     virtual ~KafkaConsumer();
  24.  
  25.     int InitCfg(const string& strBrokers, const string& strGroupID, const vector<string>& vTopics);
  26.  
  27.  
  28.     void pullMessage();
  29.  
  30.     static void RebalanceCb(rd_kafka_t *rk,
  31.         rd_kafka_resp_err_t err,
  32.         rd_kafka_topic_partition_list_t *partitions,
  33.         void *opaque);
  34.  
  35.     static void log_cb(const rd_kafka_t *rk, int level,
  36.         const char *fac, const char *buf);
  37.  
  38.     static void  EventErrorCb(rd_kafka_t *rk, int err,
  39.         const char *reason,
  40.         void *opaque);
  41.  
  42.     static int EventStatsCb(rd_kafka_t *rk,
  43.         char *json,
  44.         size_t json_len,
  45.         void *opaque);
  46.  
  47. protected:
  48.     string m_strBrokers;
  49.     string m_groupID;
  50.     vector<string> m_topicVector;
  51.     int m_partition;
  52.  
  53.     rd_kafka_t *m_pkafka;
  54. };
  55.  
  56. #endif // KAFKACONSUMER_H

kafkaConsumer.c

  1. #include "KafkaConsumer.h"
  2.  
  3. KafkaConsumer::KafkaConsumer()
  4.     : m_pkafka(nullptr)
  5. {
  6. }
  7.  
  8. int KafkaConsumer::InitCfg(const string& strBrokers, const string& strGroupID, const vector<string>& vTopics)
  9. {
  10.     m_strBrokers = strBrokers;
  11.     m_groupID = strGroupID;
  12.     m_topicVector = vTopics;
  13.  
  14.     rd_kafka_conf_t *pConf = rd_kafka_conf_new();
  15.     if(!pConf)
  16.     {
  17.         return -1;
  18.     }
  19.  
  20.     char szErr[512] = { 0};
  21.     if (rd_kafka_conf_set(pConf, "bootstrap.servers", m_strBrokers.c_str(), szErr, sizeof(szErr)) != RD_KAFKA_CONF_OK) {
  22.         fprintf(stderr, "%s\n", szErr);
  23.         rd_kafka_conf_destroy(pConf);
  24.         return -1;
  25.     }
  26.  
  27.     if (rd_kafka_conf_set(pConf, "group.id", m_groupID.c_str(), szErr, sizeof(szErr)) != RD_KAFKA_CONF_OK) {
  28.             fprintf(stderr, "%s\n", szErr);
  29.             rd_kafka_conf_destroy(pConf);
  30.             return -1;
  31.     }
  32.  
  33.     if (rd_kafka_conf_set(pConf, "statistics.interval.ms", "10000", szErr, sizeof(szErr))!= RD_KAFKA_CONF_OK) {
  34.         fprintf(stderr, "%s\n", szErr);
  35.         rd_kafka_conf_destroy(pConf);
  36.         return -1;
  37.     }
  38.     // rd_kafka_conf_set_stats_cb(pConf, &KafkaConsumer::EventStatsCb);
  39.     // 
  40.     // rd_kafka_conf_set_error_cb(pConf, &KafkaConsumer::EventErrorCb);
  41.     // rd_kafka_conf_set_log_cb(pConf, &KafkaConsumer::log_cb);
  42.     
  43.     // rd_kafka_conf_set_rebalance_cb(pConf, &KafkaConsumer::RebalanceCb);
  44.  
  45.     // topic配置
  46.     rd_kafka_topic_conf_t* pTopicConf = rd_kafka_topic_conf_new();
  47.     if (rd_kafka_topic_conf_set(pTopicConf, "auto.offset.reset", "earliest", szErr, sizeof(szErr)) != RD_KAFKA_CONF_OK) {
  48.             fprintf(stderr, "%s\n", szErr);
  49.             rd_kafka_topic_conf_destroy(pTopicConf);
  50.             return -1;
  51.     }
  52.  
  53.     // rd_kafka_conf_set_default_topic_conf(pConf, pTopicConf);
  54.  
  55.     m_pkafka = rd_kafka_new(RD_KAFKA_CONSUMER, pConf, szErr, sizeof(szErr));
  56.     if (!m_pkafka) {
  57.         fprintf(stderr, "%% Failed to create new consumer: %s\n", szErr);
  58.         return -1;
  59.     }
  60.  
  61.     rd_kafka_poll_set_consumer(m_pkafka);
  62.  
  63.     /* Convert the list of topics to a format suitable for librdkafka */
  64.     rd_kafka_topic_partition_list_t *subscription = rd_kafka_topic_partition_list_new(m_topicVector.size());
  65.     for (int i = 0 ; i < m_topicVector.size() ; i++)
  66.     {
  67.         rd_kafka_topic_partition_list_add(subscription, m_topicVector[i].c_str(), RD_KAFKA_PARTITION_UA);
  68.     }
  69.  
  70.     /* Subscribe to the list of topics */
  71.     rd_kafka_resp_err_t err = rd_kafka_subscribe(m_pkafka, subscription);
  72.     if (err) {
  73.             fprintf(stderr, "%% Failed to subscribe to %d topics: %s\n", subscription->cnt, rd_kafka_err2str(err));
  74.             rd_kafka_topic_partition_list_destroy(subscription);
  75.             rd_kafka_destroy(m_pkafka);
  76.             return -1;
  77.     }
  78.  
  79.     fprintf(stderr, "%% Subscribed to %d topic(s), " "waiting for rebalance and messages...\n", subscription->cnt);
  80.     rd_kafka_topic_partition_list_destroy(subscription);
  81.  
  82.     return 0;
  83. }
  84.  
  85. void KafkaConsumer::pullMessage()
  86. {
  87.     rd_kafka_message_t *rkm;
  88.     rkm = rd_kafka_consumer_poll(m_pkafka, 2000);
  89.     if (!rkm)
  90.     {
  91.         return ; /* Timeout: no message within 100ms,
  92.                         *  try again. This short timeout allows
  93.                         *  checking for `run` at frequent intervals.
  94.                         */
  95.     }
  96.     /* consumer_poll() will return either a proper message
  97.         * or a consumer error (rkm->err is set). */
  98.     if (rkm->err) {
  99.             /* Consumer errors are generally to be considered
  100.                 * informational as the consumer will automatically
  101.                 * try to recover from all types of errors. */
  102.             fprintf(stderr,
  103.                     "%% Consumer error: %s\n",
  104.                     rd_kafka_message_errstr(rkm));
  105.             rd_kafka_message_destroy(rkm);
  106.             return;
  107.     }
  108.  
  109.     /* Print the message value/payload. */
  110.     if (rkm->payload)
  111.     {
  112.         printf(" Value: %.*s\n", (int)rkm->len, (const char *)rkm->payload);
  113.     }
  114.     rd_kafka_message_destroy(rkm);
  115. }
  116.  
  117. void KafkaConsumer::RebalanceCb(rd_kafka_t *rk, rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t *partitions, void *opaque)
  118. {
  119.     if(partitions)
  120.     {
  121.         for (int i=0; i<partitions->cnt; i++)
  122.         {
  123.             partitions->elems[i].offset = 0;
  124.         }
  125.     }
  126.  
  127.     rd_kafka_assign(rk, partitions);
  128. }
  129.  
  130. #include <iostream>
  131. #include <stdio.h>
  132.  
  133. void KafkaConsumer::log_cb(const rd_kafka_t *rk, int level, const char *fac, const char *buf)
  134. {
  135.     cout<< "<   EventErrorCb    >" << level << "  " << fac << " " << buf <<endl;
  136. }
  137.  
  138. void KafkaConsumer::EventErrorCb(rd_kafka_t *rk, int err, const char *reason, void *opaque)
  139. {
  140.     cout<< "<   EventErrorCb    >" << err << "  " << reason <<endl;
  141. }
  142.  
  143. int KafkaConsumer::EventStatsCb(rd_kafka_t *rk, char *json, size_t json_len, void *opaque)
  144. {
  145.     cout<< "<   EventStatsCb    >" << json <<endl;
  146.     return 0;
  147. }
  148.  
  149. KafkaConsumer::~KafkaConsumer()
  150. {
  151.     rd_kafka_consumer_close(m_pkafka);
  152.     /* Destroy the consumer */
  153.     rd_kafka_destroy(m_pkafka);
  154.     m_pkafka = nullptr;
  155. }

libkafkacpp C++消费者接口实现：

  1. #include "KafkaConsumer.h"
  2.  
  3. KafkaConsumer::KafkaConsumer(const std::string& brokers, const std::string& groupID,
  4.                              const std::vector<std::string>& topics, int partition)
  5. {
  6.     m_brokers = brokers;
  7.     m_groupID = groupID;
  8.     m_topicVector = topics;
  9.     m_partition = partition;
  10.  
  11.     std::string errorStr;
  12.     RdKafka::Conf::ConfResult errorCode;
  13.     m_config = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  14.  
  15.     m_event_cb = new ConsumerEventCb;
  16.     errorCode = m_config->set("event_cb", m_event_cb, errorStr);
  17.     if(errorCode != RdKafka::Conf::CONF_OK)
  18.     {
  19.         std::cout << "Conf set failed: " << errorStr << std::endl;
  20.     }
  21.  
  22.     m_rebalance_cb = new ConsumerRebalanceCb;
  23.     errorCode = m_config->set("rebalance_cb", m_rebalance_cb, errorStr);
  24.     if(errorCode != RdKafka::Conf::CONF_OK)
  25.     {
  26.         std::cout << "Conf set failed: " << errorStr << std::endl;
  27.     }
  28.  
  29.     errorCode = m_config->set("enable.partition.eof", "false", errorStr);
  30.     if(errorCode != RdKafka::Conf::CONF_OK)
  31.     {
  32.         std::cout << "Conf set failed: " << errorStr << std::endl;
  33.     }
  34.  
  35.     errorCode = m_config->set("group.id", m_groupID, errorStr);
  36.     if(errorCode != RdKafka::Conf::CONF_OK)
  37.     {
  38.         std::cout << "Conf set failed: " << errorStr << std::endl;
  39.     }
  40.     errorCode = m_config->set("bootstrap.servers", m_brokers, errorStr);
  41.     if(errorCode != RdKafka::Conf::CONF_OK)
  42.     {
  43.         std::cout << "Conf set failed: " << errorStr << std::endl;
  44.     }
  45.     errorCode = m_config->set("max.partition.fetch.bytes", "1024000", errorStr);
  46.     if(errorCode != RdKafka::Conf::CONF_OK)
  47.     {
  48.         std::cout << "Conf set failed: " << errorStr << std::endl;
  49.     }
  50.  
  51.     // partition.assignment.strategy  range,roundrobin
  52.  
  53.     m_topicConfig = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
  54.     // 获取最新的消息数据
  55.     errorCode = m_topicConfig->set("auto.offset.reset", "latest", errorStr);
  56.     if(errorCode != RdKafka::Conf::CONF_OK)
  57.     {
  58.         std::cout << "Topic Conf set failed: " << errorStr << std::endl;
  59.     }
  60.     errorCode = m_config->set("default_topic_conf", m_topicConfig, errorStr);
  61.     if(errorCode != RdKafka::Conf::CONF_OK)
  62.     {
  63.         std::cout << "Conf set failed: " << errorStr << std::endl;
  64.     }
  65.     m_consumer = RdKafka::KafkaConsumer::create(m_config, errorStr);
  66.     if(m_consumer == NULL)
  67.     {
  68.         std::cout << "Create KafkaConsumer failed: " << errorStr << std::endl;
  69.     }
  70.     std::cout << "Created consumer " << m_consumer->name() << std::endl;
  71. }
  72.  
  73. void msg_consume(RdKafka::Message* msg, void* opaque)
  74. {
  75.     switch (msg->err())
  76.     {
  77.     case RdKafka::ERR__TIMED_OUT:
  78.         std::cerr << "Consumer error: " << msg->errstr() << std::endl;
  79.         break;
  80.     case RdKafka::ERR_NO_ERROR:     // 有消息进来
  81.         std::cout << " Message in " << msg->topic_name() << " ["
  82.                   << msg->partition() << "] at offset " << msg->offset()
  83.                   << " key: " << msg->key() << " payload: "
  84.                   << (char*)msg->payload() << std::endl;
  85.         break;
  86.     default:
  87.         std::cerr << "Consumer error: " << msg->errstr() << std::endl;
  88.         break;
  89.     }
  90. }
  91.  
  92. void KafkaConsumer::pullMessage()
  93. {
  94.     // 订阅Topic
  95.     RdKafka::ErrorCode errorCode = m_consumer->subscribe(m_topicVector);
  96.     if (errorCode != RdKafka::ERR_NO_ERROR)
  97.     {
  98.         std::cout << "subscribe failed: " << RdKafka::err2str(errorCode) << std::endl;
  99.     }
  100.     // 消费消息
  101.     while(true)
  102.     {
  103.         RdKafka::Message *msg = m_consumer->consume(1000);
  104.         msg_consume(msg, NULL);
  105.         delete msg;
  106.     }
  107. }
  108.  
  109. KafkaConsumer::~KafkaConsumer()
  110. {
  111.     m_consumer->close();
  112.     delete m_config;
  113.     delete m_topicConfig;
  114.     delete m_consumer;
  115.     delete m_event_cb;
  116.     delete m_rebalance_cb;
  117.  
  118. }

libkafkacpp C++生产者接口实现：

  1. #include "KafkaProducer.h"
  2.  
  3. KafkaProducer::KafkaProducer(const std::string& brokers, const std::string& topic, int partition)
  4. {
  5.     m_brokers = brokers;
  6.     m_topicStr = topic;
  7.     m_partition = partition;
  8.     // 创建Kafka Conf对象
  9.     m_config = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  10.     if(m_config == NULL)
  11.     {
  12.         std::cout << "Create RdKafka Conf failed." << std::endl;
  13.     }
  14.     // 创建Topic Conf对象
  15.     m_topicConfig = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
  16.     if(m_topicConfig == NULL)
  17.     {
  18.         std::cout << "Create RdKafka Topic Conf failed." << std::endl;
  19.     }
  20.     // 设置Broker属性
  21.     RdKafka::Conf::ConfResult errCode;
  22.     m_dr_cb = new ProducerDeliveryReportCb;
  23.     std::string errorStr;
  24.     errCode = m_config->set("dr_cb", m_dr_cb, errorStr);
  25.     if(errCode != RdKafka::Conf::CONF_OK)
  26.     {
  27.         std::cout << "Conf set failed:" << errorStr << std::endl;
  28.     }
  29.     m_event_cb = new ProducerEventCb;
  30.     errCode = m_config->set("event_cb", m_event_cb, errorStr);
  31.     if(errCode != RdKafka::Conf::CONF_OK)
  32.     {
  33.         std::cout << "Conf set failed:" << errorStr << std::endl;
  34.     }
  35.  
  36.     m_partitioner_cb = new HashPartitionerCb;
  37.     errCode = m_topicConfig->set("partitioner_cb", m_partitioner_cb, errorStr);
  38.     if(errCode != RdKafka::Conf::CONF_OK)
  39.     {
  40.         std::cout << "Conf set failed:" << errorStr << std::endl;
  41.     }
  42.  
  43.     errCode = m_config->set("statistics.interval.ms", "10000", errorStr);
  44.     if(errCode != RdKafka::Conf::CONF_OK)
  45.     {
  46.         std::cout << "Conf set failed:" << errorStr << std::endl;
  47.     }
  48.  
  49.     errCode = m_config->set("message.max.bytes", "10240000", errorStr);
  50.     if(errCode != RdKafka::Conf::CONF_OK)
  51.     {
  52.         std::cout << "Conf set failed:" << errorStr << std::endl;
  53.     }
  54.     errCode = m_config->set("bootstrap.servers", m_brokers, errorStr);
  55.     if(errCode != RdKafka::Conf::CONF_OK)
  56.     {
  57.         std::cout << "Conf set failed:" << errorStr << std::endl;
  58.     }
  59.  
  60.     // request.required.acks  0=Broker does not send any response/ack to client, 
  61.     //                        -1 or all=Broker will block until message is committed by all in sync replicas (ISRs)
  62.  
  63.     // 创建Producer
  64.     m_producer = RdKafka::Producer::create(m_config, errorStr);
  65.     if(m_producer == NULL)
  66.     {
  67.         std::cout << "Create Producer failed:" << errorStr << std::endl;
  68.     }
  69.     // 创建Topic对象
  70.     m_topic = RdKafka::Topic::create(m_producer, m_topicStr, m_topicConfig, errorStr);
  71.     if(m_topic == NULL)
  72.     {
  73.         std::cout << "Create Topic failed:" << errorStr << std::endl;
  74.     }
  75. }
  76.  
  77. void KafkaProducer::pushMessage(const std::string& str, const std::string& key)
  78. {
  79.     int32_t len = str.length();
  80.     void* payload = const_cast<void*>(static_cast<const void*>(str.data()));
  81.     RdKafka::ErrorCode errorCode = m_producer->produce(m_topic, RdKafka::Topic::PARTITION_UA,
  82.                                    RdKafka::Producer::RK_MSG_COPY,
  83.                                    payload, len, &key, NULL);
  84.     m_producer->poll(0);
  85.     if (errorCode != RdKafka::ERR_NO_ERROR)
  86.     {
  87.         std::cerr << "Produce failed: " << RdKafka::err2str(errorCode) << std::endl;
  88.         if(errorCode ==  RdKafka::ERR__QUEUE_FULL)
  89.         {
  90.             m_producer->poll(100);
  91.         }
  92.     }
  93. }
  94.  
  95. KafkaProducer::~KafkaProducer()
  96. {
  97.     while (m_producer->outq_len() > 0)
  98.     {
  99.         std::cerr << "Waiting for " << m_producer->outq_len() << std::endl;
  100.         m_producer->flush(5000);
  101.     }
  102.     delete m_config;
  103.     delete m_topicConfig;
  104.     delete m_topic;
  105.     delete m_producer;
  106.     delete m_dr_cb;
  107.     delete m_event_cb;
  108.     delete m_partitioner_cb;
  109. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/酷酷是懒虫/article/detail/894046
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号