当前位置: 首页 > news >正文

Centos 7下使用C++使用Rdkafka库实现生产者消费者

1. 了解 Kafka

Apache Kafka 是一个分布式流处理平台,核心功能包括:

  • 发布/订阅消息系统:解耦生产者和消费者

  • 分布式存储:持久化、容错的消息存储

  • 流处理:实时处理数据流

核心概念

概念说明
BrokerKafka 集群中的单个服务器节点
Topic消息的逻辑分类(如:user_activity
PartitionTopic 的分区(并行处理单位),消息按顺序存储
Producer向 Topic 发布消息的客户端
Consumer订阅 Topic 并处理消息的客户端
Consumer Group多个消费者协同消费同一 Topic(每个分区只被组内一个消费者消费)
Offset消息在分区中的唯一位置标识

2. 了解 rdkafka

rdkafka 是 Kafka 的 C/C++ 客户端库,提供:

  • 高性能生产/消费 API(支持 C/C++/Python 等)

  • 特性:

    • 异步/同步发送模式

    • 自动负载均衡

    • 消息压缩(gzip, snappy, lz4)

    • SASL 认证

    • 精确一次语义(EOS)

  • 开源地址:edenhill/librdkafka

3. 代码实现

以下是使用 librdkafka 的 C++ 接口操作 Kafka 的生产者和消费者完整实现:

生产者代码 (producer.cpp)
#include <iostream>
#include <string>
#include <sstream>
#include <librdkafka/rdkafkacpp.h>class ProducerDeliveryReportCb : public RdKafka::DeliveryReportCb {
public:void dr_cb(RdKafka::Message &message) {if (message.err()) {std::cerr << "消息发送失败: " << message.errstr() << std::endl;} else {std::cout << "消息发送成功: " << message.topic_name() << " [" << message.partition() << "] @ " << message.offset() << std::endl;}}
};int main() {// 1. 创建配置对象RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);std::string errstr;// 2. 设置配置参数if (conf->set("bootstrap.servers", "localhost:9092", errstr) != RdKafka::Conf::CONF_OK) {std::cerr << "配置错误: " << errstr << std::endl;return 1;}// 设置消息确认模式 (all = 最高可靠性)if (conf->set("acks", "all", errstr) != RdKafka::Conf::CONF_OK) {std::cerr << "配置错误: " << errstr << std::endl;return 1;}// 3. 创建生产者实例ProducerDeliveryReportCb delivery_cb;if (conf->set("dr_cb", &delivery_cb, errstr) != RdKafka::Conf::CONF_OK) {std::cerr << "配置回调错误: " << errstr << std::endl;return 1;}RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);if (!producer) {std::cerr << "创建生产者失败: " << errstr << std::endl;return 1;}delete conf;// 4. 创建Topic对象RdKafka::Conf *tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);RdKafka::Topic *topic = RdKafka::Topic::create(producer,"cpp_test_topic",tconf,errstr);if (!topic) {std::cerr << "创建Topic失败: " << errstr << std::endl;delete tconf;return 1;}delete tconf;// 5. 生产消息for (int i = 0; i < 10; ++i) {std::string key = "key-" + std::to_string(i);std::string payload = "Message #" + std::to_string(i);// 发送消息RdKafka::ErrorCode resp = producer->produce(topic, RdKafka::Topic::PARTITION_UA, // 自动分区分配RdKafka::Producer::RK_MSG_COPY,const_cast<char*>(payload.c_str()), payload.size(),const_cast<char*>(key.c_str()), key.size(),NULL);if (resp != RdKafka::ERR_NO_ERROR) {std::cerr << "生产消息失败: " << RdKafka::err2str(resp) << std::endl;} else {std::cout << "已发送: " << payload << std::endl;}// 处理事件队列producer->poll(0);}// 6. 等待所有消息完成发送while (producer->outq_len() > 0) {std::cout << "等待发送队列: " << producer->outq_len() << std::endl;producer->poll(100);}// 7. 清理资源delete topic;delete producer;return 0;
}
消费者代码 (consumer.cpp)
#include <iostream>
#include <string>
#include <csignal>
#include <vector>
#include <librdkafka/rdkafkacpp.h>bool running = true;void stop(int sig) {running = false;
}class ConsumerEventCb : public RdKafka::EventCb {
public:void event_cb(RdKafka::Event &event) {switch (event.type()) {case RdKafka::Event::EVENT_ERROR:std::cerr << "错误: " << RdKafka::err2str(event.err()) << std::endl;break;case RdKafka::Event::EVENT_LOG:std::cout << "日志: " << event.str() << std::endl;break;default:std::cout << "事件: " << event.type() << ": " << event.str() << std::endl;break;}}
};int main() {// 注册信号处理signal(SIGINT, stop);signal(SIGTERM, stop);// 1. 创建配置对象RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);std::string errstr;// 2. 设置配置参数if (conf->set("bootstrap.servers", "localhost:9092", errstr) != RdKafka::Conf::CONF_OK) {std::cerr << "配置错误: " << errstr << std::endl;return 1;}// 设置消费组if (conf->set("group.id", "cpp_consumer_group", errstr) != RdKafka::Conf::CONF_OK) {std::cerr << "配置错误: " << errstr << std::endl;return 1;}// 从最早的消息开始消费if (conf->set("auto.offset.reset", "earliest", errstr) != RdKafka::Conf::CONF_OK) {std::cerr << "配置错误: " << errstr << std::endl;return 1;}// 3. 设置事件回调ConsumerEventCb event_cb;if (conf->set("event_cb", &event_cb, errstr) != RdKafka::Conf::CONF_OK) {std::cerr << "设置回调失败: " << errstr << std::endl;return 1;}// 4. 创建消费者实例RdKafka::KafkaConsumer *consumer = RdKafka::KafkaConsumer::create(conf, errstr);if (!consumer) {std::cerr << "创建消费者失败: " << errstr << std::endl;return 1;}delete conf;// 5. 订阅Topicstd::vector<std::string> topics;topics.push_back("cpp_test_topic");RdKafka::ErrorCode resp = consumer->subscribe(topics);if (resp != RdKafka::ERR_NO_ERROR) {std::cerr << "订阅失败: " << RdKafka::err2str(resp) << std::endl;return 1;}// 6. 消费消息while (running) {// 等待消息 (1000ms超时)RdKafka::Message *msg = consumer->consume(1000);switch (msg->err()) {case RdKafka::ERR__TIMED_OUT:break;  // 超时继续case RdKafka::ERR_NO_ERROR:// 成功消费到消息std::cout << "收到消息: "<< "主题: " << msg->topic_name() << " | 分区: [" << msg->partition() << "]"<< " | 偏移量: " << msg->offset() << std::endl;if (msg->key()) {std::cout << "键: " << *msg->key() << " => ";}std::cout << "值: " << static_cast<const char*>(msg->payload()) << std::endl;break;default:std::cerr << "消费错误: " << msg->errstr() << std::endl;break;}// 手动提交偏移量consumer->commitAsync(msg);delete msg;}// 7. 关闭消费者consumer->close();delete consumer;return 0;
}
编译运行

# 编译生产者
g++ -o producer producer.cpp -lrdkafka++ -lrdkafka -lpthread -lz -ldl# 编译消费者
g++ -o consumer consumer.cpp -lrdkafka++ -lrdkafka -lpthread -lz -ldl

http://www.dtcms.com/a/272946.html

相关文章:

  • 【LeetCode 热题 100】19. 删除链表的倒数第 N 个结点——双指针+哨兵
  • 学习 Flutter (一)
  • html的outline: none;
  • C++STL-deque
  • 1. COLA-DDD的实战
  • 【基础架构】——软件系统复杂度的来源(低成本、安全、规模)
  • 告别卡顿与慢响应!现代 Web 应用性能优化:从前端渲染到后端算法的全面提速指南
  • IDEA运行Spring项目报错:java: 警告: 源发行版 17 需要目标发行版 17,java: 无效的目标发行版: 17
  • Cargo.toml 配置详解
  • 【科研绘图系列】R语言探索生物多样性与地理分布的可视化之旅
  • 网安-解决pikachu-rce乱码问题
  • 访问Windows服务器备份SQL SERVER数据库
  • (C++)任务管理系统(文件存储)(正式版)(迭代器)(list列表基础教程)(STL基础知识)
  • x86交叉编译ros 工程给jetson nano运行
  • Rust and the Linux Kernel
  • Sophix、Tinker 和 Robust 三大主流 Android 热修复框架的详细对比
  • windows10 安装docker到H盘
  • Linux 服务器挖矿病毒深度处理与防护指南
  • 使用Docker将Python项目部署到云端的完整指南
  • Web 会话认证方案详解:原理、流程与安全实践
  • Variables
  • 分库分表之实战-sharding-JDBC分库分表执行流程原理剖析
  • ubantu问题手册
  • 大数据学习5:网站访问日志分析
  • 力扣hot100速通(7.9)|49.字母异位词分组 128.最长连续序列 283.移动零 11.盛最多水的容器 42.接雨水
  • 观成科技:基于自监督学习技术的恶意加密流量检测方案
  • CRMEB Pro版前端环境配置指南
  • AT9850B北斗双频导航定位芯片简介
  • 基于大数据的电力系统故障诊断技术研究
  • 华为鸿蒙HarmonyOpenEye项目:开眼App的鸿蒙实现之旅