詳解C/C++如何發(fā)送與接收Kafka消息
更新時間：2024年07月17日 09:53:20 作者：周先生FullStack
系統(tǒng)之間通信方式很多如：系統(tǒng)之間調(diào)用（http/rpc等）,異步間接調(diào)用如發(fā)送消息、公共存儲等,算法工程為C/C++工程,本文將介紹如何在C/C++中如何發(fā)送與接收Kakfa消息（包含：Kafka的SASL認(rèn)證方式）,并提供了詳細(xì)的源碼和講解,需要的朋友可以參考下
一、背景
在實際工程中，難免會遇到不通系統(tǒng)之間通信，如何進(jìn)行系統(tǒng)之間通信呢？(作為一個“全棧工程師”，必須要解決它！)。
系統(tǒng)之間通信方式很多如：系統(tǒng)之間調(diào)用（http/rpc等），異步間接調(diào)用如發(fā)送消息、公共存儲等。目前，本人從事的項目中遇到web業(yè)務(wù)工程（Java）依賴與算法工程(C++) 處理的視頻/圖片分類與標(biāo)記結(jié)果。兩個系統(tǒng)之前數(shù)據(jù)通信采用了kafka消息方式。
算法工程為C/C++工程，本文將介紹如何在C/C++中如何發(fā)送與接收Kakfa消息（包含：Kafka的SASL認(rèn)證方式），并提供了詳細(xì)的源碼和講解。（至于Java中如何發(fā)送與接收Kakfa消息如有需要，可留言或私聊?。?/p>
二、環(huán)境依賴安裝
# 下載librdkafka git clone https://github.com/edenhill/librdkafka.git # 編譯 cd librdkafka ./configure --prefix=/usr/local # 安裝 sudo make install # 驗證：查看/usr/local/lib目錄下是否有l(wèi)ibrdkafka文件 ls /usr/local/lib | grep kafka
三、編寫kakfa生產(chǎn)者消費者
3.1 生產(chǎn)者
#include <rdkafka.h> // 包含C API頭文件
#include <iostream>
#include <cstring>
#include <cerrno>
int main() {
const char *brokers = "xx.xx.xx.xx:7091"; // Kafka broker地址
const char *topic_name = "kafka_msg_topic_test";
const char *payload = "Hello, Kafka from librdkafka!";
size_t len = strlen(payload);
// 創(chuàng)建配置對象
rd_kafka_conf_t *conf = rd_kafka_conf_new();
if (!conf) {
std::cerr << "Failed to create configuration object: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
return 1;
}
// 設(shè)置broker地址
if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers, NULL, 0) != RD_KAFKA_CONF_OK) {
std::cerr << "Failed to set bootstrap.servers: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_conf_destroy(conf);
return 1;
}
// 創(chuàng)建生產(chǎn)者實例
rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, NULL, 0);
if (!rk) {
std::cerr << "Failed to create producer: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_conf_destroy(conf);
return 1;
}
// 創(chuàng)建topic句柄（可選，但推薦）
rd_kafka_topic_t *rkt = rd_kafka_topic_new(rk, topic_name, NULL);
if (!rkt) {
std::cerr << "Failed to create topic handle: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_destroy(rk);
// rd_kafka_conf_destroy(conf);
return 1;
}
// 發(fā)送消息
int32_t partition = RD_KAFKA_PARTITION_UA; // 自動選擇分區(qū)
int err = rd_kafka_produce(rkt, partition, RD_KAFKA_MSG_F_COPY, const_cast<char *>(payload), len, NULL, 0, NULL);
if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
std::cerr << "Failed to produce to topic " << topic_name << ": " << err << std::endl;
} else {
std::cout << "Produced " << len << " bytes to topic " << topic_name << std::endl;
}
// 等待所有消息發(fā)送完成（可選，但推薦）
// 在實際生產(chǎn)代碼中，您可能需要更復(fù)雜的邏輯來處理消息的發(fā)送和確認(rèn)
int msgs_sent = 0;
while (rd_kafka_outq_len(rk) > 0) {
rd_kafka_poll(rk, 100); // 輪詢Kafka隊列，直到所有消息都發(fā)送出去
msgs_sent += rd_kafka_outq_len(rk);
}
// 銷毀topic句柄
rd_kafka_topic_destroy(rkt);
// 銷毀生產(chǎn)者實例
rd_kafka_destroy(rk);
// 銷毀配置對象
// rd_kafka_conf_destroy(conf);
return 0;
}3.2 消費者
#include <rdkafka.h>
#include <iostream>
#include <cerrno>
#include <cstring>
#include <cstdlib>
void error_cb(rd_kafka_t *rk, int err, const char *reason, void *opaque) {
// 錯誤處理回調(diào)
std::cerr << "Kafka error: " << err << ": " << reason << std::endl;
}
int main() {
std::cerr << "start " << std::endl;
const char *brokers = "xx.xx.xx.xx:7091"; // Kafka broker地址
const char *group_id = "kafka_msg_topic_test"; // 消費者組ID
const char *topic_name = "kafka_msg_topic_test"; // Kafka topic名稱
// 創(chuàng)建配置對象
rd_kafka_conf_t *conf = rd_kafka_conf_new();
if (!conf) {
std::cerr << "Failed to create configuration object: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
return 1;
}
// 設(shè)置broker地址
if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers, NULL, 0) != RD_KAFKA_CONF_OK) {
std::cerr << "Failed to set bootstrap.servers: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_conf_destroy(conf);
return 1;
}
// 設(shè)置消費者組ID
if (rd_kafka_conf_set(conf, "group.id", group_id, NULL, 0) != RD_KAFKA_CONF_OK) {
std::cerr << "Failed to set group.id: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_conf_destroy(conf);
return 1;
}
// 設(shè)置錯誤處理回調(diào)（可選）
rd_kafka_conf_set_error_cb(conf, error_cb);
// 創(chuàng)建消費者實例
rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, NULL, 0);
if (!rk) {
std::cerr << "Failed to create consumer: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
return 1;
}
// 創(chuàng)建一個topic分區(qū)列表
rd_kafka_topic_partition_list_t *topics = rd_kafka_topic_partition_list_new(1);
if (!topics) {
std::cerr << "Failed to create topic partition list: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_destroy(rk);
return 1;
}
// 添加topic到分區(qū)列表
if (!rd_kafka_topic_partition_list_add(topics, topic_name, RD_KAFKA_PARTITION_UA)) {
std::cerr << "Failed to add topic to partition list: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_topic_partition_list_destroy(topics);
rd_kafka_destroy(rk);
return 1;
}
// 訂閱topic
rd_kafka_resp_err_t err = rd_kafka_subscribe(rk, topics);
if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
std::cerr << "Failed to subscribe to topic: " << rd_kafka_err2str(err) << std::endl;
rd_kafka_topic_partition_list_destroy(topics);
rd_kafka_destroy(rk);
return 1;
}
// 銷毀分區(qū)列表（訂閱后不再需要）
rd_kafka_topic_partition_list_destroy(topics);
// 輪詢消息
while (true) {
rd_kafka_message_t *rkmessage;
rkmessage = rd_kafka_consumer_poll(rk, 1000); // 等待1秒以獲取消息
if (rkmessage == NULL) {
// 沒有消息或者超時
continue;
}
if (rkmessage->err) {
// 處理錯誤
if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
// 消息流的末尾
std::cout << "End of partition event" << std::endl;
} else {
// 打印錯誤并退出
std::cerr << "Kafka consumer error: " << rd_kafka_message_errstr(rkmessage) << std::endl;
break;
}
} else {
// 處理消息
std::cout << "Received message at offset " << rkmessage->offset
<< " from partition " << rkmessage->partition
<< " with key \"" << rkmessage->key << "\" and payload size "<< rkmessage->len
<< " value :" <<(char *)rkmessage->payload
<< std::endl;
// 如果需要，可以在這里處理消息內(nèi)容
// 例如，使用rkmessage->payload()獲取消息內(nèi)容
// 釋放消息
rd_kafka_message_destroy(rkmessage);
}
}
// 清理
rd_kafka_destroy(rk);
return 0;
}3.3 編譯運行
3.3.1 編譯生產(chǎn)者消費者
g++ -o send_kafka SendKakfaMessage.cpp -I/usr/local/include/librdkafka -lrdkafka++ -lrdkafka -lpthread
g++ -o receive_kafka ReceiveKafkaMessage.cpp -I/usr/local/include/librdkafka -lrdkafka++ -lrdkafka -lpthread
3.3.2 運行驗證
執(zhí)行時，若出現(xiàn)錯誤： error while loading shared libraries: librdkafka++.so.1: cannot open shared object file: No such file or directory
則需要執(zhí)行下面環(huán)境變量配置：
export LD_LIBRARY_PATH=/usr/local/lib:$LD_LIBRARY_PATH
生產(chǎn)者：發(fā)送消息
消費者：接收消息
3.4 SASL認(rèn)證kakfa
下面是，支持sasl認(rèn)證的kakka生產(chǎn)者完整代碼
#include <rdkafka.h>
#include <iostream>
#include <cstring>
#include <cerrno>
int main(int argc, char *argv[]) {
const char *brokers = "xx.xx.xx.xx:8092"; // Kafka broker地址
const char *username = "xxx";
const char *password = "xxx";
const char *topic_name = "kafka_msg_test_sasl";
const char *payload = "Hello, Kafka from librdkafka! sasl";
size_t len = strlen(payload);
// 初始化配置
rd_kafka_conf_t *conf = rd_kafka_conf_new();
if (!conf)
{
std::cerr << "Failed to create configuration object: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
return 1;
}
char errstr[512]; // 聲明一個足夠大的字符數(shù)組來存儲錯誤信息
// 設(shè)置SASL相關(guān)的配置
if (rd_kafka_conf_set(conf, "security.protocol", "SASL_PLAINTEXT", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
std::cerr << "Failed to set security.protocol: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_conf_destroy(conf);
return 1;
}
if (rd_kafka_conf_set(conf, "sasl.mechanisms", "PLAIN", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
std::cerr << "Failed to set sasl.mechanisms: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_conf_destroy(conf);
return 1;
}
if (rd_kafka_conf_set(conf, "sasl.username", username, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
std::cerr << "Failed to set sasl.username: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_conf_destroy(conf);
return 1;
}
if (rd_kafka_conf_set(conf, "sasl.password", password, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
std::cerr << "Failed to set sasl.password: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_conf_destroy(conf);
return 1;
}
if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
std::cerr << "Failed to set bootstrap.servers: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_conf_destroy(conf);
return 1;
}
// 檢查配置是否設(shè)置成功
if (rd_kafka_conf_set(conf, "security.protocol", "SASL_PLAINTEXT", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
std::cerr << "Failed to set configuration: " << errstr << std::endl;
return 1;
}
// 創(chuàng)建producer實例
rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
if (!rk) {
std::cerr << "Failed to create new producer: " << errstr << std::endl;
return 1;
}
// 創(chuàng)建topic句柄（可選，但推薦）
rd_kafka_topic_t *rkt = rd_kafka_topic_new(rk, topic_name, NULL);
if (!rkt)
{
std::cerr << "Failed to create topic handle: " << rd_kafka_err2str(rd_kafka_last_error()) << std::endl;
rd_kafka_destroy(rk);
// rd_kafka_conf_destroy(conf);
return 1;
}
// 發(fā)送消息
int32_t partition = RD_KAFKA_PARTITION_UA; // 自動選擇分區(qū)
int err = rd_kafka_produce(rkt, partition, RD_KAFKA_MSG_F_COPY, const_cast<char *>(payload), len, NULL, 0, NULL);
if (err != RD_KAFKA_RESP_ERR_NO_ERROR)
{
std::cerr << "Failed to produce to topic " << topic_name << ": " << err << std::endl;
}
else
{
std::cout << "Produced " << len << " bytes to topic " << topic_name << std::endl;
}
// 等待所有消息發(fā)送完成（可選，但推薦）
// 在實際生產(chǎn)代碼中，您可能需要更復(fù)雜的邏輯來處理消息的發(fā)送和確認(rèn)
int msgs_sent = 0;
while (rd_kafka_outq_len(rk) > 0)
{
rd_kafka_poll(rk, 100); // 輪詢Kafka隊列，直到所有消息都發(fā)送出去
msgs_sent += rd_kafka_outq_len(rk);
}
// 銷毀topic句柄
rd_kafka_topic_destroy(rkt);
// 清理資源
rd_kafka_destroy(rk);
return 0;
}在kafka map 管理界面中查看發(fā)送效果如下：
3.5 結(jié)束語
到此這篇關(guān)于詳解C/C++如何發(fā)送與接收Kafka消息的文章就介紹到這了,更多相關(guān)C/C++發(fā)送與接收Kafka內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了C++封裝IATHOOK類的實現(xiàn)方法,對IAT的HOOK實例進(jìn)行了封裝,非常具有實用價值,需要的朋友可以參考下2014-10-10
這篇文章主要介紹了 c語言排序之歸并排序，歸并就是把兩個或多個序列合并，文中通過示例代碼介紹的非常詳細(xì)。對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2022-05-05
這篇文章主要介紹了C語言內(nèi)存泄露很嚴(yán)重的解決方案，預(yù)防內(nèi)存泄漏問題有多種方法，比如加強(qiáng)代碼檢視、工具檢測和內(nèi)存測試等，下面文章總結(jié)內(nèi)容需要的小伙伴可以參考一下2022-05-05
C語言 深入探究動態(tài)規(guī)劃之區(qū)間DP
這幾天在做有關(guān)dp的題，看到一個石子合并的問題，本來以為是個貪心，后來仔細(xì)一想壓根不是貪心。貪心算法的思路是每次都取最大的，然而石子合并問題有個限制條件就是每次只能取相鄰的，這就決定了它不是個貪心2022-04-04
C++高級數(shù)據(jù)結(jié)構(gòu)之二叉查找樹
這篇文章主要介紹了C++高級數(shù)據(jù)結(jié)構(gòu)之二叉查找樹,文章圍繞主題展開詳細(xì)的內(nèi)容介紹，具有一定的參考價值，需要的小伙伴可以參考一下2022-05-05
這篇文章主要介紹了C語言實現(xiàn)的順序表功能,結(jié)合完整實例形式分析了C語言順序表的創(chuàng)建、添加、刪除、排序、合并等相關(guān)操作技巧,需要的朋友可以參考下2018-04-04
Visual Studio 2019 如何新建 Win32項目的方法步驟
這篇文章主要介紹了Visual Studio 2019 如何新建 Win32項目的方法步驟，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-03-03
這篇文章主要給大家總結(jié)介紹了C語言中一些“不受限制”的字符串函數(shù)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03
這篇文章主要介紹了C語言中的malloc的使用,包括用其動態(tài)申請二維數(shù)組等功能,需要的朋友可以參考下2015-08-08
現(xiàn)在正做一個接口，通過不同的連接字符串操作不同的數(shù)據(jù)庫。要用到mysql數(shù)據(jù)庫。通過網(wǎng)上的一些資料和自己的摸索，大致清楚了C++連接mysql的方法?？梢酝ㄟ^2種方法實現(xiàn)。第一種方法是利用ADO連接，第二種方法是利用mysql自己的api函數(shù)進(jìn)行連接。今天主要來講解下使用API2016-07-07