Windows10配置C++版本的Kafka，并进行发布和订阅测试

配置的情况为：Release x64下的情况
完整项目：https://gitee.com/jiajingong/kafka-publisher
1、首先下载相应的库文件（.lib，.dll）

参考链接：
GitHub - eStreamSoftware/delphi-kafka
GitHub - cloader/KafkaCPP-win32-dll: KafkaCpp-win32-dll
2、新建一个新的下令行C++工程

建完工程后，选择Release x64，并在天生中执行重新天生解决方案，这样会在项目目次下天生x64/Release文件夹
3、通过VS2017配置附加库目次和附加依赖项

所有的.lib、.dll等库文件均在下图x64/Release目次下

附加依赖项到场：librdkafka.lib;librdkafkacpp.lib，如下图：

4、发布端：

将主函数的CPP文件改为：

1. #include <iostream>
2. #include <thread>
3. #include "rdkafkacpp.h"
5. int main()
6. {
7.         std::string brokers = "172.18.4.96:9092";
8.         std::string errorStr;
9.         RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
10.         RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
11.         if (!conf) {
12.                 std::cout << "Create RdKafka Conf failed" << std::endl;
13.                 return -1;
14.         }
16.         conf->set("message.max.bytes", "10240000", errorStr); //最大字节数
17.         conf->set("replica.fetch.max.bytes", "20485760", errorStr);
18.         conf->set("bootstrap.servers", brokers, errorStr);
20.         RdKafka::Producer *producer = RdKafka::Producer::create(conf, errorStr);
21.         if (!producer) {
22.                 std::cout << "Create Producer failed" << std::endl;
23.                 return -1;
24.         }
25.         //创建Topic
26.         RdKafka::Topic *topic = RdKafka::Topic::create(producer, "koala-stqf-03", tconf, errorStr);
27.         if (!topic) {
28.                 std::cout << "Create Topic failed" << std::endl;
29.         }
30.         int count = 0;
31.         while (true)
32.         {   //发送消息
33.                 RdKafka::ErrorCode resCode = producer->produce(topic, RdKafka::Topic::PARTITION_UA, RdKafka::Producer::RK_MSG_COPY, (char *)"123456789", 10, nullptr, nullptr);
34.                 std::cout << "Count:" << count << ",has publish:" << (char *)"123456789" << std::endl;
35.                 if (resCode != RdKafka::ERR_NO_ERROR) {
36.                         std::cerr << "Produce failed: " << RdKafka::err2str(resCode) << std::endl;
37.                 }
38.                 count += 1;
39.                 std::this_thread::sleep_for(std::chrono::seconds(1));
40.         }
42.         delete conf;
43.         delete tconf;
44.         delete topic;
45.         delete producer;
47.         RdKafka::wait_destroyed(5000);
48.         return 0;
49. }

复制代码

5、订阅端

新建一个同样的订阅端工程，同样将主函数的代码改为：

1. #include "rdkafkacpp.h"
2. #include <chrono>
3. #include <time.h>
4. #include <sstream>
5. #include <iomanip>
6. #include <iostream>
7. #include <algorithm>
8. #include <iterator>
10. void consume_cb(RdKafka::Message &message, void *opaque)
11. {
12.         switch (message.err()) {
13.         case RdKafka::ERR__TIMED_OUT:
14.                 std::cout << "RdKafka::ERR__TIMED_OUT" << std::endl;
15.                 break;
16.         case RdKafka::ERR_NO_ERROR:
17.                 /* Real message */
19.                 RdKafka::MessageTimestamp ts;
20.                 ts = message.timestamp();
21.                 if (ts.type != RdKafka::MessageTimestamp::MSG_TIMESTAMP_NOT_AVAILABLE) {
22.                         std::string timeprefix;
23.                         if (ts.type == RdKafka::MessageTimestamp::MSG_TIMESTAMP_CREATE_TIME) {
24.                                 timeprefix = "created time";
25.                         }
26.                         else if (ts.type == RdKafka::MessageTimestamp::MSG_TIMESTAMP_LOG_APPEND_TIME) {
27.                                 timeprefix = "log append time";
28.                         }
31.                         unsigned long long milli = ts.timestamp + (unsigned long long)8 * 60 * 60 * 1000;//此处转化为东八区北京时间，如果是其它时区需要按需求修改
32.                         auto mTime = std::chrono::milliseconds(milli);
33.                         auto tp = std::chrono::time_point<std::chrono::system_clock, std::chrono::milliseconds>(mTime);
34.                         auto tt = std::chrono::system_clock::to_time_t(tp);
35.                         tm timeinfo;
36.                         ::gmtime_s(&timeinfo, &tt);
38.                         //char s[60]{ 0 };
39.                         //::sprintf(s, "%04d-%02d-%02d %02d:%02d:%02d", timeinfo.tm_year + 1900, timeinfo.tm_mon + 1, timeinfo.tm_mday, timeinfo.tm_hour, timeinfo.tm_min, timeinfo.tm_sec);
40.                         // std::cout << timeinfo.tm_year + 1900 << "-" << timeinfo.tm_mon + 1 << "-" << timeinfo.tm_mday << " " << timeinfo.tm_hour << ":" << timeinfo.tm_min << ":" << timeinfo.tm_sec << std::endl;
41. #if 0
42.                         std::stringstream ss;
43.                         std::string dateStr;
45.                         ss << timeinfo.tm_year + 1900 << "-"
46.                                 << timeinfo.tm_mon + 1 << "-"
47.                                 << timeinfo.tm_mday;
48.                         ss >> dateStr;
50.                         ss.clear();
51.                         ss << timeinfo.tm_hour << ":"
52.                                 << timeinfo.tm_min << ":"
53.                                 << timeinfo.tm_sec;
54.                         std::string timeStr;
55.                         ss >> timeStr;
57.                         std::string dateTimeStr;
58.                         dateTimeStr += dateStr;
59.                         dateTimeStr.push_back(' ');
60.                         dateTimeStr += timeStr;
61. #endif // 0
63.                         //std::cout << "TimeStamp" << timeprefix << " " << s << std::endl;
64.                         std::cout << "TimeStamp   " << timeinfo.tm_year + 1900 << "-" << timeinfo.tm_mon + 1 << "-" << timeinfo.tm_mday << " " << timeinfo.tm_hour << ":" << timeinfo.tm_min << ":" << timeinfo.tm_sec << std::endl;
65.                 }
67.                 std::cout << message.topic_name() << " offset" << message.offset() << "  partion " << message.partition() << " message: " << reinterpret_cast<char*>(message.payload()) << std::endl;
68.                 break;
70.         case RdKafka::ERR__PARTITION_EOF:
71.                 /* Last message */
72.                 std::cout << "EOF reached for" << std::endl;
73.                 break;
75.         case RdKafka::ERR__UNKNOWN_TOPIC:
76.         case RdKafka::ERR__UNKNOWN_PARTITION:
77.                 std::cout << "Consume failed: " << message.errstr();
78.                 break;
80.         default:
81.                 /* Errors */
82.                 std::cout << "Consume failed: " << message.errstr();
83.                 break;
84.         }
85. }
86. int main()
87. {
88.         std::string brokers = "172.18.4.96:9092";
89.         std::string errstr;
90.         std::vector<std::string> topics{ "koala-stqf-03",
91.                 "klai-seim-alert-koala-test-03"
92.         };
93.         std::string group_id = "whl-consumer-group";
95.         RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
96.         if (conf->set("group.id", group_id, errstr)) {
97.                 std::cout << errstr << std::endl;
98.                 return -1;
99.         }
101.         conf->set("bootstrap.servers", brokers, errstr);
102.         conf->set("max.partition.fetch.bytes", "1024000", errstr);
103.         //conf->set("enable-auto-commit", "true", errstr);
104.         RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);
105.         tconf->set("auto.offset.reset", "latest", errstr);
106.         conf->set("default_topic_conf", tconf, errstr);
108.         RdKafka::KafkaConsumer *m_consumer = RdKafka::KafkaConsumer::create(conf, errstr);
109.         if (!m_consumer) {
110.                 std::cout << "failed to create consumer " << errstr << std::endl;
111.                 return -1;
112.         }
114. #if 0 //从上一次消费结束的位置开始消费
115.         RdKafka::ErrorCode err = m_consumer->subscribe(topics);
116.         if (err != RdKafka::ERR_NO_ERROR) {
117.                 std::cout << RdKafka::err2str(err) << std::endl;
118.                 return -1;
119.         }
120. #else //指定每个topic的每个分区开始消费的位置
122.         //基本思路为先获取server端的状态信息，将与订阅相关的topic找出来，根据分区，创建TopicPartion；最后使用assign消费
123.         RdKafka::Metadata *metadataMap{ nullptr };
124.         RdKafka::ErrorCode err = m_consumer->metadata(true, nullptr, &metadataMap, 2000);
125.         if (err != RdKafka::ERR_NO_ERROR) {
126.                 std::cout << RdKafka::err2str(err) << std::endl;
127.         }
128.         const RdKafka::Metadata::TopicMetadataVector *topicList = metadataMap->topics();
129.         std::cout << "broker topic size: " << topicList->size() << std::endl;
130.         RdKafka::Metadata::TopicMetadataVector subTopicMetaVec;
131.         std::copy_if(topicList->begin(), topicList->end(), std::back_inserter(subTopicMetaVec), [&topics](const RdKafka::TopicMetadata* data) {
132.                 return std::find_if(topics.begin(), topics.end(), [data](const std::string &tname) {return data->topic() == tname; }) != topics.end();
133.         });
134.         std::vector<RdKafka::TopicPartition*> topicpartions;
135.         std::for_each(subTopicMetaVec.begin(), subTopicMetaVec.end(), [&topicpartions](const RdKafka::TopicMetadata* data) {
136.                 auto parVec = data->partitions();
137.                 std::for_each(parVec->begin(), parVec->end(), [&](const RdKafka::PartitionMetadata *value) {
138.                         std::cout << data->topic() << " has partion: " << value->id() << " Leader is : " << value->leader() << std::endl;
139.                         topicpartions.push_back(RdKafka::TopicPartition::create(data->topic(), value->id(), RdKafka::Topic::OFFSET_END));
140.                 });
141.         });
142.         m_consumer->assign(topicpartions);
143. #endif // 0
144.         RdKafka::ErrorCode errccc = m_consumer->subscribe(topics);
145.         if (errccc != RdKafka::ERR_NO_ERROR) {
146.                 std::cout << RdKafka::err2str(errccc) << std::endl;
147.                 return -1;
148.         }
150.         while (true)
151.         {
152.                 RdKafka::Message *msg = m_consumer->consume(6000);
153.                 consume_cb(*msg, nullptr); //消息一条消息
154.                 delete msg;
155.         }
156.         return 0;
157. }

复制代码

6、发布 订阅展示：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。