librdkafka的简单使用
文章目录
- 摘要
- kafka是什么
- 安装环境
- librdkafka的简单使用
- 生产者
- 消费者
摘要
本文是Getting Started with Apache Kafka and C/C++的中文版, kafka的hello world程序。
本文完整代码见仓库,这里只列出producer/consumer的代码
kafka是什么
本节来源:Kafka - 维基百科,自由的百科全书、Kafka入门简介 - 知乎
首先我们得知道什么是Kafka。
Kafka是一个分布式流处理系统,流处理系统使它可以像消息队列一样publish或者subscribe消息,分布式提供了容错性,并发处理消息的机制。
kafka有以下一些基本概念:
- Producer - 消息生产者,就是向kafka broker发消息的客户端。
- Consumer - 消息消费者,是消息的使用方,负责消费Kafka服务器上的消息。
- Topic - 主题,由用户定义并配置在Kafka服务器,用于建立Producer和Consumer之间的订阅关系。生产者发送消息到指定的Topic下,消息者从这个Topic下消费消息。
- Partition - 消息分区,一个topic可以分为多个 partition,每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id(offset)。
- Broker - 一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。
- Consumer Group - 消费者分组,用于归组同类消费者。每个consumer属于一个特定的consumer group,多个消费者可以共同消费一个Topic下的消息,每个消费者消费其中的部分消息,这些消费者就组成了一个分组,拥有同一个分组名称,通常也被称为消费者集群。
- Offset - 消息在partition中的偏移量。每一条消息在partition都有唯一的偏移量,消费者可以指定偏移量来指定要消费的消息。
安装环境
上一节,kafka的概念看着比较简单,发布-订阅/生产-消费的模型。
为了可以调用Kafka的C/C++ API, 需要先安装环境。
# almlinux8
# dnf search kafka
# dnf install librdkafka-devel# dnf search glib
# dnf install glib2-devel# ubuntu22
# 开发库
➜ apt install librdkafka-dev libglib2.0-dev# 安装docker环境
➜ apt install docker.io docker-compose# 本地安装 Kafka
## ref: https://docs.confluent.io/confluent-cli/current/install.html#cp
➜ wget -qO - https://packages.confluent.io/confluent-cli/deb/archive.key | sudo apt-key add
➜ add-apt-repository "deb https://packages.confluent.io/confluent-cli/deb stable main"
➜ apt install confluent-cli## 启动kafka
## usage: https://docs.confluent.io/confluent-cli/current/command-reference/local/kafka/confluent_local_kafka_start.html
## error: https://stackoverflow.com/questions/63776518/error-2-matches-found-based-on-name-network-nameofservice-default-is-ambiguo
## error:https://stackoverflow.com/questions/77985757/kafka-in-docker-using-confluent-cli-doesnt-work➜ whereis confluent
confluent: /usr/bin/confluent➜ export CONFLUENT_HOME=/usr/bin/confluent# 我执行下面命令后,没有看到Plaintext Ports信息
➜ confluent local kafka start# 停止,然后重新启动,管用了
➜ confluent local kafka stop
➜ confluent local kafka startThe local commands are intended for a single-node development environment only, NOT for production usage. See more: https://docs.confluent.io/current/cli/index.htmlPulling from confluentinc/confluent-local
Digest: sha256:ad62269bf4766820c298f7581cf872a49f46a11dbaebcccb4fd2e71049288c5b
Status: Image is up to date for confluentinc/confluent-local:7.6.0
+-----------------+-------+
| Kafka REST Port | 8082 |
| Plaintext Ports | 43465 |
+-----------------+-------+
Started Confluent Local containers "8d72d911a4".
To continue your Confluent Local experience, run `confluent local kafka topic create <topic>` and `confluent local kafka topic produce <topic>`.# Create a new topic, purchases, which you will use to produce and consume events.
➜ confluent local kafka topic create purchases
Created topic "purchases".
librdkafka的简单使用
confluenceinc/librdkafka是Apache Kafka协议的 C 库实现 ,提供生产者、消费者和管理客户端。
下面运行的程序来自:Apache Kafka and C/C++ - Getting Started Tutorial
代码中kafka的API可以查询:librdkafka: librdkafka documentation
代码中使用了glib库,日常开发我不会使用这个库,因为感觉比较冷,它的API可查询:GLib – 2.0: Automatic Cleanup
生产者
总体逻辑:
- 从配置文件中加载配置
- 创建生产者
- 生产者发送消息
#include <glib.h>
#include <librdkafka/rdkafka.h>#include "common.c"#define ARR_SIZE(arr) ( sizeof((arr)) / sizeof((arr[0])) )/* Optional per-message delivery callback (triggered by poll() or flush())* when a message has been successfully delivered or permanently* failed delivery (after retries).*/
static void dr_msg_cb (rd_kafka_t *kafka_handle,const rd_kafka_message_t *rkmessage,void *opaque) {if (rkmessage->err) {g_error("Message delivery failed: %s", rd_kafka_err2str(rkmessage->err));}
}int main (int argc, char **argv) {rd_kafka_t *producer;rd_kafka_conf_t *conf;char errstr[512];// Parse the command line.if (argc != 2) {g_error("Usage: %s <config.ini>", argv[0]);return 1;}// Parse the configuration.// See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.mdconst char *config_file = argv[1];g_autoptr(GError) error = NULL;g_autoptr(GKeyFile) key_file = g_key_file_new();if (!g_key_file_load_from_file (key_file, config_file, G_KEY_FILE_NONE, &error)) {g_error ("Error loading config file: %s", error->message);return 1;}// Load the relevant configuration sections.conf = rd_kafka_conf_new();load_config_group(conf, key_file, "default");// Install a delivery-error callback.rd_kafka_conf_set_dr_msg_cb(conf, dr_msg_cb);// Create the Producer instance.producer = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));if (!producer) {g_error("Failed to create new producer: %s", errstr);return 1;}// Configuration object is now owned, and freed, by the rd_kafka_t instance.conf = NULL;// Produce data by selecting random values from these lists.int message_count = 10;const char *topic = "purchases";const char *user_ids[6] = {"eabara", "jsmith", "sgarcia", "jbernard", "htanaka", "awalther"};const char *products[5] = {"book", "alarm clock", "t-shirts", "gift card", "batteries"};for (int i = 0; i < message_count; i++) {const char *key = user_ids[random() % ARR_SIZE(user_ids)];const char *value = products[random() % ARR_SIZE(products)];size_t key_len = strlen(key);size_t value_len = strlen(value);rd_kafka_resp_err_t err;err = rd_kafka_producev(producer,RD_KAFKA_V_TOPIC(topic),RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY),RD_KAFKA_V_KEY((void*)key, key_len),RD_KAFKA_V_VALUE((void*)value, value_len),RD_KAFKA_V_OPAQUE(NULL),RD_KAFKA_V_END);if (err) {g_error("Failed to produce to topic %s: %s", topic, rd_kafka_err2str(err));return 1;} else {g_message("Produced event to topic %s: key = %12s value = %12s", topic, key, value);}rd_kafka_poll(producer, 0);}// Block until the messages are all sent.g_message("Flushing final messages..");rd_kafka_flush(producer, 10 * 1000);if (rd_kafka_outq_len(producer) > 0) {g_error("%d message(s) were not delivered", rd_kafka_outq_len(producer));}g_message("%d events were produced to topic %s.", message_count, topic);rd_kafka_destroy(producer);return 0;
}
消费者
总体逻辑:
- 从配置文件中加载配置
- 创建消费者
- 订阅topic
- 轮询消费者的消息
#include <glib.h>
#include <librdkafka/rdkafka.h>#include "common.c"static volatile sig_atomic_t run = 1;/*** @brief Signal termination of program*/
static void stop(int sig) { run = 0; }int main(int argc, char **argv) {rd_kafka_t *consumer;rd_kafka_conf_t *conf;rd_kafka_resp_err_t err;char errstr[512];// Parse the command line.if (argc != 2) {g_error("Usage: %s <config.ini>", argv[0]);return 1;}// Parse the configuration.// See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.mdconst char *config_file = argv[1];g_autoptr(GError) error = NULL;g_autoptr(GKeyFile) key_file = g_key_file_new();if (!g_key_file_load_from_file(key_file, config_file, G_KEY_FILE_NONE,&error)) {g_error("Error loading config file: %s", error->message);return 1;}// Load the relevant configuration sections.conf = rd_kafka_conf_new();load_config_group(conf, key_file, "default");load_config_group(conf, key_file, "consumer");// Create the Consumer instance.consumer = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));if (!consumer) {g_error("Failed to create new consumer: %s", errstr);return 1;}rd_kafka_poll_set_consumer(consumer);// Configuration object is now owned, and freed, by the rd_kafka_t instance.conf = NULL;// Convert the list of topics to a format suitable for librdkafka.const char *topic = "purchases";rd_kafka_topic_partition_list_t *subscription =rd_kafka_topic_partition_list_new(1);rd_kafka_topic_partition_list_add(subscription, topic, RD_KAFKA_PARTITION_UA);// Subscribe to the list of topics.err = rd_kafka_subscribe(consumer, subscription);if (err) {g_error("Failed to subscribe to %d topics: %s", subscription->cnt,rd_kafka_err2str(err));rd_kafka_topic_partition_list_destroy(subscription);rd_kafka_destroy(consumer);return 1;}rd_kafka_topic_partition_list_destroy(subscription);// Install a signal handler for clean shutdown.signal(SIGINT, stop);// Start polling for messages.while (run) {rd_kafka_message_t *consumer_message;consumer_message = rd_kafka_consumer_poll(consumer, 500);if (!consumer_message) {g_message("Waiting...");continue;}if (consumer_message->err) {if (consumer_message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {/* We can ignore this error - it just means we've read* everything and are waiting for more data.*/} else {g_message("Consumer error: %s",rd_kafka_message_errstr(consumer_message));return 1;}} else {g_message("Consumed event from topic %s: key = %.*s value = %s",rd_kafka_topic_name(consumer_message->rkt),(int)consumer_message->key_len, (char *)consumer_message->key,(char *)consumer_message->payload);}// Free the message when we're done.rd_kafka_message_destroy(consumer_message);}// Close the consumer: commit final offsets and leave the group.g_message("Closing consumer");rd_kafka_consumer_close(consumer);// Destroy the consumer.rd_kafka_destroy(consumer);return 0;
}
相关文章:

librdkafka的简单使用
文章目录 摘要kafka是什么安装环境librdkafka的简单使用生产者消费者 摘要 本文是Getting Started with Apache Kafka and C/C的中文版, kafka的hello world程序。 本文完整代码见仓库,这里只列出producer/consumer的代码 kafka是什么 本节来源&#…...

【iOS ARKit】播放3D音频
3D音频 在前面系列中,我们了解如何定位追踪用户(实际是定位用户的移动设备)的位置与方向,然后通过摄像机的投影矩阵将虚拟物体投影到用户移动设备屏幕。如果用户移动了,则通过VIO 和 IMU更新用户的位置与方向信息&…...
ES学习日记(四)-------插件head安装和一些配套插件下载
前言 接上节,第三方插件选择了时间久,功能丰富,长得丑的head,head 插件在ES 5版本以前开箱即用非常简单,ES 5版本以后需要运行在node环境下,所以我们要先准备一下环境 一.安装Git 不装了,明儿再说,看会儿手机准备下班!!!!!!!!!...
flask+uwsgi+云服务器 部署服务端
参考:使用uwsgi部署flask 报错 “找不到Python应用程序,请检查启动日志以查找错误” 或者: no python application found, check your startup logs for errors debug 过程:查到Python uWSGI 安装配置 里面说,先写测…...

linux学习之路 -- 普通用户添加进sudoer列表
在Linux系统里,很多的操作普通用户是不能执行的,所以我们需要对普通用户进行提权操作,可我们会发现,一开始没有配置的话,是无法的提权操作的,下面我将介绍普通用户该如何配置sudoer列表。 首先以root 的身…...

【分类评估指标,精确率,召回率,】from sklearn.metrics import classification_report
from: https://zhuanlan.zhihu.com/p/368196647 多分类 from sklearn.metrics import classification_report y_true [0, 1, 2, 2, 2] y_pred [0, 0, 2, 2, 1] target_names [class 0, class 1, class 2] # print(classification_report(y_true, y_pred, targe…...

element-ui autocomplete 组件源码分享
紧接着 input 组件的源码,分享带输入建议的 autocomplete 组件,在 element-ui 官方文档上,没有这个组件的 api 目录,它的 api 是和 input 组件的 api 在一起的,看完源码之后发现,源码当中 autocomplete 组件…...
视觉SLAM理论与实践的学习链接汇总
仅供学习,在此感谢所有乐于分享知识的大佬们~ 一、 ORB_SLAM理论 视觉SLAM 前端 后端 回环 建图 1、 前端视觉里程计 1.1 特征点法 一文带你搞懂相机内参外参(Intrinsics & Extrinsics)-知乎 VSLAM 笔记——我们如何通过图像来计算位姿的变化ÿ…...

极光笔记|极光消息推送服务的云原生实践
摘要 极光始终秉承“以开发者为中心”的战略导向,极光推送(JPush)是国内领先的消息推送服务。极光推送(JPush)本质上是一种软件付费应用程序,结合当前主流云厂商基础施设,逐渐演进成了云上SaaS…...
高效八股文背诵方法
往往到了找工作高峰期,经常会出现八股文很多 难以背诵 的苦恼,下面在下结合情况,列举了几点自认为可以的背诵方法: 1. **大声朗读**: - 对于Java核心概念和重要理论,先大声朗读,这不仅可以帮…...

Codeforces Round 841 (Div. 2) C. Even Subarrays
题目 思路: #include <bits/stdc.h> using namespace std; #define int long long #define pb push_back #define fi first #define se second #define lson p << 1 #define rson p << 1 | 1 const int maxn 1e6 5, inf 1e9, maxm 4e4 5; co…...

用 SpringBoot+Redis 解决海量重复提交问题
1前言 在实际的开发项目中,一个对外暴露的接口往往会面临很多次请求,我们来解释一下幂等的概念:任意多次执行所产生的影响均与一次执行的影响相同。按照这个含义,最终的含义就是 对数据库的影响只能是一次性的,不能重复处理。如何…...

前端基础知识html
一.基础标签 1.<h1>-<h6>:定义标题,h最大,h最小 2.<font>:定义文本的字体,尺寸,颜色 3.<b>:定义粗体文本 4.<i>:定义斜体文本 5.<u>:定义文本下…...

网络原理-传输层-UDP报文结构
本文介绍UDP报文 有很多友友搞不清楚UDP报文的详细结构还有TCP的详细结构,所以专门分开来讲 以免弄混. 首先我们先看一下整个UDP结构,让大家有一个全方面的认识 下面我们来详细解释UDP报 16位源端口号(本机):就是2字节大小,16个二进制位. 16位目的端口号(目的机):也是2字节…...
TCP/IP参考模型(四层及其解析)
文章目录 1、什么是TCP/IP2、四层协议2.1 应用层(应用程序协议)2.2 传输层(源端口↔️目的端口)2.3 网络层(主机↔️主机)2.4 网络接口层(主机↔️网络层) 总结 1、什么是TCP/IP TC…...

2024第六届环境科学与可再生能源国际会议能源 (ESRE 2024) 即将召开!
2024第六届环境科学与可再生能源国际会议 能源 (ESRE 2024) 即将举行 2024 年 6 月 28 日至 30 日在德国法兰克福举行。ESRE 2024 年 旨在为研究人员、从业人员和专业人士提供一个论坛 从工业界、学术界和政府到研究和 发展,环境科学领域的专…...
CentOS配置docker外部访问
CoreOS 官方文档提供的方法 官方文档:https://coreos.com/os/docs/latest/customizing-docker.html 新建 /etc/systemd/system/docker-tcp.socket 文件 [Unit] DescriptionDocker Socket for the API[Socket] # ListenStream127.0.0.1:2375 ListenStre…...
面试前端八股文十问十答第二期
面试前端八股文十问十答第二期 作者:程序员小白条,个人博客 相信看了本文后,对你的面试是有一定帮助的!关注专栏后就能收到持续更新! ⭐点赞⭐收藏⭐不迷路!⭐ 1)从输入URL到页面加载的全过程…...

【漏洞复现】大华综合安防监控管理平台 Digital Surveillance System系统存在RCE漏洞
免责声明:文章来源互联网收集整理,请勿利用文章内的相关技术从事非法测试,由于传播、利用此文所提供的信息或者工具而造成的任何直接或者间接的后果及损失,均由使用者本人负责,所产生的一切不良后果与文章作者无关。该…...

ssm网上订餐管理系统开发mysql数据库web结构java编程计算机网页源码eclipse项目采用线性算法
一、源码特点 ssm 网上订餐管理系统是一套完善的信息系统,结合springMVC框架完成本系统,对理解JSP java编程开发语言有帮助系统采用SSM框架(MVC模式开发),系统具有完整的源代码和数据库,系统主要采用B/S模…...

XCTF-web-easyupload
试了试php,php7,pht,phtml等,都没有用 尝试.user.ini 抓包修改将.user.ini修改为jpg图片 在上传一个123.jpg 用蚁剑连接,得到flag...

中南大学无人机智能体的全面评估!BEDI:用于评估无人机上具身智能体的综合性基准测试
作者:Mingning Guo, Mengwei Wu, Jiarun He, Shaoxian Li, Haifeng Li, Chao Tao单位:中南大学地球科学与信息物理学院论文标题:BEDI: A Comprehensive Benchmark for Evaluating Embodied Agents on UAVs论文链接:https://arxiv.…...
【解密LSTM、GRU如何解决传统RNN梯度消失问题】
解密LSTM与GRU:如何让RNN变得更聪明? 在深度学习的世界里,循环神经网络(RNN)以其卓越的序列数据处理能力广泛应用于自然语言处理、时间序列预测等领域。然而,传统RNN存在的一个严重问题——梯度消失&#…...

STM32标准库-DMA直接存储器存取
文章目录 一、DMA1.1简介1.2存储器映像1.3DMA框图1.4DMA基本结构1.5DMA请求1.6数据宽度与对齐1.7数据转运DMA1.8ADC扫描模式DMA 二、数据转运DMA2.1接线图2.2代码2.3相关API 一、DMA 1.1简介 DMA(Direct Memory Access)直接存储器存取 DMA可以提供外设…...

Mac软件卸载指南,简单易懂!
刚和Adobe分手,它却总在Library里给你写"回忆录"?卸载的Final Cut Pro像电子幽灵般阴魂不散?总是会有残留文件,别慌!这份Mac软件卸载指南,将用最硬核的方式教你"数字分手术"࿰…...

全志A40i android7.1 调试信息打印串口由uart0改为uart3
一,概述 1. 目的 将调试信息打印串口由uart0改为uart3。 2. 版本信息 Uboot版本:2014.07; Kernel版本:Linux-3.10; 二,Uboot 1. sys_config.fex改动 使能uart3(TX:PH00 RX:PH01),并让boo…...
【Java学习笔记】BigInteger 和 BigDecimal 类
BigInteger 和 BigDecimal 类 二者共有的常见方法 方法功能add加subtract减multiply乘divide除 注意点:传参类型必须是类对象 一、BigInteger 1. 作用:适合保存比较大的整型数 2. 使用说明 创建BigInteger对象 传入字符串 3. 代码示例 import j…...
虚拟电厂发展三大趋势:市场化、技术主导、车网互联
市场化:从政策驱动到多元盈利 政策全面赋能 2025年4月,国家发改委、能源局发布《关于加快推进虚拟电厂发展的指导意见》,首次明确虚拟电厂为“独立市场主体”,提出硬性目标:2027年全国调节能力≥2000万千瓦࿰…...
Vite中定义@软链接
在webpack中可以直接通过符号表示src路径,但是vite中默认不可以。 如何实现: vite中提供了resolve.alias:通过别名在指向一个具体的路径 在vite.config.js中 import { join } from pathexport default defineConfig({plugins: [vue()],//…...

TSN交换机正在重构工业网络,PROFINET和EtherCAT会被取代吗?
在工业自动化持续演进的今天,通信网络的角色正变得愈发关键。 2025年6月6日,为期三天的华南国际工业博览会在深圳国际会展中心(宝安)圆满落幕。作为国内工业通信领域的技术型企业,光路科技(Fiberroad&…...