Kafka 的消息格式:了解消息结构与序列化
Kafka 作为一款高性能的消息中间件系统,其消息格式对于消息的生产、传输和消费起着至关重要的作用。本篇博客将深入讨论 Kafka 的消息格式,包括消息的结构、序列化与反序列化,以及一些常用的消息格式选项。通过更丰富的示例代码和深入的解析,希望能够帮助大家更好地理解 Kafka 消息的内部机制。
1. Kafka 消息结构
Kafka 的消息结构由消息头、消息键、消息值和时间戳等组成。下面是一个典型的 Kafka 消息结构:
----------------------------------------------------------------------------------------------
| Message Header | Key | Value | Timestamp | Optional Headers |
----------------------------------------------------------------------------------------------
1.1 消息头
消息头包含一些元数据信息,例如消息的大小、压缩信息等。消息头的结构可能会根据 Kafka 版本和配置而有所不同。
1.2 消息键与消息值
-
消息键(Key): 用于标识消息的唯一性,通常用于分区和查找消息。
-
消息值(Value): 包含实际的消息内容。
1.3 时间戳
时间戳表示消息的产生时间,有两种类型:
-
创建时间戳: 表示消息被创建的时间。
-
LogAppendTime 时间戳: 表示消息被追加到日志的时间。
2. 消息的序列化与反序列化
Kafka 中的消息在生产者发送和消费者接收时需要进行序列化和反序列化。这是因为 Kafka 是以字节流的形式存储和传输消息的,而实际的消息内容可能是各种不同的数据类型。以下是一些常用的序列化器和反序列化器:
2.1 字符串序列化器
// 生产者端
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "Hello, Kafka!");// 消费者端
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
records.forEach(record -> {System.out.printf("Consumed record with key %s and value %s%n", record.key(), record.value());
});
2.2 Avro 序列化器
Avro 是一种高性能且紧凑的二进制序列化格式,适用于复杂数据结构的消息。
// 生产者端
GenericRecord avroRecord = new GenericData.Record(schema);
avroRecord.put("field1", "value1");
avroRecord.put("field2", 42);
ProducerRecord<String, GenericRecord> record = new ProducerRecord<>("my-topic", "key", avroRecord);// 消费者端
ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
records.forEach(record -> {GenericRecord value = record.value();System.out.printf("Consumed record with key %s and value %s%n", record.key(), value);
});
2.3 JSON 序列化器
// 生产者端
JsonNode jsonNode = objectMapper.createObjectNode();
((ObjectNode) jsonNode).put("field1", "value1");
((ObjectNode) jsonNode).put("field2", 42);
ProducerRecord<String, JsonNode> record = new ProducerRecord<>("my-topic", "key", jsonNode);// 消费者端
ConsumerRecords<String, JsonNode> records = consumer.poll(Duration.ofMillis(100));
records.forEach(record -> {JsonNode value = record.value();System.out.printf("Consumed record with key %s and value %s%n", record.key(), value);
});
3. 自定义消息格式
在某些情况下,你可能需要定义自己的消息格式。Kafka 提供了 ByteArraySerializer
和 ByteArrayDeserializer
,允许你将消息以字节数组的形式发送和接收,从而实现自定义的序列化和反序列化逻辑。
// 生产者端
byte[] customMessageBytes = serializeCustomMessage(customMessage);
ProducerRecord<String, byte[]> record = new ProducerRecord<>("my-topic", "key", customMessageBytes);// 消费者端
ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
records.forEach(record -> {byte[] value = record.value();CustomMessage customMessage = deserializeCustomMessage(value);System.out.printf("Consumed record with key %s and value %s%n", record.key(), customMessage);
});
4. 消息的压缩与解压
Kafka 支持消息的压缩,以减小网络传输的开销。以下是一些常用的压缩选项:
// 生产者端
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip");
Producer<String, String> producer = new KafkaProducer<>(props);// 消费者端
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
Consumer<String, String> consumer = new KafkaConsumer<>(props);
5. 消息的版本控制与兼容性
在实际应用中,系统的演进和变化是不可避免的。因此,考虑到消息的版本控制和兼容性是非常重要的。以下是一些相关的注意事项和最佳实践:
5.1 消息的演进
-
向后兼容性: 新版本的消费者能够处理旧版本的消息。
-
向前兼容性: 旧版本的消费者能够处理新版本的消息。
5.2 Schema Registry
Schema Registry 是一个用于存储和管理 Avro、JSON 等消息格式的架构的中心化服务。通过使用 Schema Registry,可以更好地管理消息的演进,并确保向前和向后的兼容性。
// 配置 Schema Registry 地址
props.put("schema.registry.url", "http://schema-registry:8081");
6. 消息的认证与加密
Kafka 提供了安全性特性,包括消息的认证和加密。以下是一些相关的配置选项:
6.1 SSL 加密通信
// 生产者端
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");// 消费者端
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
6.2 认证配置
// 生产者端
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";");// 消费者端
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";");
7. 消息的追踪与监控
追踪和监控是保障系统稳定性和性能的重要手段。以下是一些常用的追踪和监控工具:
7.1 JMX 监控
Kafka 提供了 JMX 接口,可以通过 JConsole 或其他 JMX 客户端进行监控。
7.2 Kafka Manager
Kafka Manager 是一款开源的 Kafka 集群管理和监控工具,提供了直观的 Web 界面。
7.3 Prometheus 和 Grafana
使用 Prometheus 进行指标采集,结合 Grafana 进行可视化展示,可以更全面地监控 Kafka 集群的性能和健康状况。
总结
在深入探讨Kafka消息格式、版本控制、安全性和监控等关键主题后,对构建高效、灵活的消息系统有了更为全面的认识。了解消息结构、序列化与反序列化、自定义消息格式,以及消息的压缩与解压,是确保消息传递的基础。随后,版本控制与兼容性的重要性得到了强调,Schema Registry成为管理Avro、JSON等消息格式的利器。在保障消息传递安全方面,SSL加密通信和认证配置提供了可靠的手段。最后,通过JMX监控、Kafka Manager、以及Prometheus和Grafana的运用,能够实时追踪和监控Kafka集群的健康状态。
这篇文章旨在为大家提供全方位的Kafka消息系统知识,使其能够在实际应用中根据业务需求构建稳健、高效的消息处理系统。深入理解这些关键概念,将有助于确保消息系统的可维护性、稳定性和安全性,为实际业务场景中的挑战提供可行的解决方案。继续关注更多Kafka相关的技术内容,将使大家能够不断深化对消息系统的认识,应对日益复杂的数据处理需求。
相关文章:
Kafka 的消息格式:了解消息结构与序列化
Kafka 作为一款高性能的消息中间件系统,其消息格式对于消息的生产、传输和消费起着至关重要的作用。本篇博客将深入讨论 Kafka 的消息格式,包括消息的结构、序列化与反序列化,以及一些常用的消息格式选项。通过更丰富的示例代码和深入的解析&…...
装箱 Box 数据类型
装箱是最简单直接的一种智能指针,它的类型是Box<T>。装箱使我们可以把数据存储到堆上,并在栈上保留一个指向堆数据的指针。装箱操作常常被用于下面的场景: 当你拥有一个无法在编译时确定大小的类型,但又想使用这个类型的值…...
多传感器融合SLAM在自动驾驶方向的初步探索的记录
1. VIO的不可观问题 现有的VIO都是解决的六自由度的问题, 但是对于行驶在路面上的车来说, 通常情况下不会有roll与z方向的自由度, 而且车体模型限制了不可能有纯yaw的变换. 同时由于IMU在Z轴上与roll, pitch上激励不足, 会导致IMU在初始化过程中尺度不准以及重力方向估计错误,…...
ffmpeg与opencv-python处理视频
安装 opencv pip install opencv-pythonFFmpeg 1.下载 FFmpeg 访问FFmpeg官方网站。选择 “Windows builds from gyan.dev” 链接,这会带您到一个包含最新版本 FFmpeg Windows 构建的页面。选择一个适合您系统的版本(例如,32位或64位&…...
java 操作git
实现功能:借助jgit实现拉取文件,并返回文件路径清单 <!-- 依赖库 版本号有自行选择,只是需要注意支持的jdk版本即可,我使用的是jdk1.8--> <dependency><groupId>org.eclipse.jgit</groupId><artif…...
Linux 导入、导出 MySQL 数据库命令
一、导出数据库 1、导出完整数据:表结构数据 mysqldump -u用户名 -p 数据库名 > 数据库名.sql 举例:以下命令可以导出 abc 数据库的数据和表结构 /usr/local/mysql/bin/mysqldump -uroot -p abc > abc.sql2、只导出表结构 mysqldump -u用户名 -p…...
华为数通---BFD多跳检测示例
定义 双向转发检测BFD(Bidirectional Forwarding Detection)是一种全网统一的检测机制,用于快速检测、监控网络中链路或者IP路由的转发连通状况。 目的 为了减小设备故障对业务的影响,提高网络的可靠性,网络设备需要…...
AWS 日志分析工具
当您的网络资源托管在 AWS 中时,需要定期监控您的 AWS CloudTrail 日志、Amazon S3 服务器日志和 AWS ELB 日志等云日志,以降低任何潜在的安全风险、识别严重错误并确保满足所有合规性法规。 什么是 Amazon S3 Amazon Simple Storage Serviceÿ…...
gitLab 和Idea分支合并
以下二选1即可完成分支合并建议第一种简单有效 Idea合并方式 切换到被合并的分支,如我想把0701的内容合并到dev,切换到dev分支,然后再点击merge然后选择要合并的分支,即可,此时git上的代码没有更新只是把代码合到本地需要pull才…...
关于 mapboxgl 的常用方法及效果
给地图标记点 实现效果 /*** 在地图上添加标记点* point: [lng, lat]* color: #83f7a0*/addMarkerOnMap(point, color #83f7a0) {const marker new mapboxgl.Marker({draggable: false,color: color,}).setLngLat(point).addTo(this.map);this.markersList.push(marker);},…...
C语言——二级指针
指针变量也是变量,是变量就有地址,那么指针变量的地址存放在哪里?——这就是二期指针 int a 10;int *pa &a;int **ppa &pa;//a的地址存放在pa中,pa的地址存放在ppa中。 //pa是一级指针,ppa是二级指针。 对…...
股市复苏中的明懿金汇:抓住新机遇
2023年对于明懿金汇来说是充满挑战与机遇的一年。面对复杂多变的市场环境,明懿金汇展现了其对市场趋势的敏锐洞察和卓越的策略适应能力。以下是该公司在2023年的主要投资策略和市场适应方式的详细分析。 随着2023年中国股市迎来反弹,明懿金汇迅速调整了…...
Spacemesh、Kaspa和Chia的全面对比!
当今区块链领域,PoST(Proof of Space and Time)共识算法引领着一股新的技术浪潮。在这个热潮下,Chia项目作为PoST共识机制的经典项目,和目前算力赛道备受瞩目的Kaspa项目,都是不可忽视的存在。虽然这两个项…...
【HTML语法】
HTML语法 1. HTML语法1.1 HTML编辑器1.2 HTML模板1.3 标签示例1.4 常见的HTML标签1.51.61.71.81.91.101.11 学习网站:https://www.runoob.com/html/html-tutorial.html 1. HTML语法 HTML(全称 Hypertext Markup Language,超文本标记语言&…...
ROS报错:RLException:Invalid roslaunch XML Syntax: mismatched tag:
运行roslaunch文件提示: RLException:Invalid roslaunch XML Syntax: mismatched tag: line 45, column 2 The traceback for the exception was written to the log file. j 解决办法: line45 行多了标签:</node> 另外…...
C语言实现快速排序
完整代码: #include<stdio.h>//用第一个元素将待排序序列划分成左右两个部分,返回排序后low的位置,即枢轴的位置 int partition(int arr[],int low,int high){//让待排序序列中的第一个元素成为基准int pivotarr[low];//lowhigh代表一…...
ChatGPT对于当今的社会或科技发展有何重要性?
ChatGPT对于当今社会和科技发展的重要性在于: 促进社交交流:ChatGPT可以为人们提供全天候的在线聊天服务,连接人与人之间的沟通交流,改善社交沟通方式。 提高有效性和效率:人们可以通过ChatGPT获得快速和精确的信息&a…...
宝塔是可以切换mongodb版本的
在软件商店,搜索monggodb,点击设置。点击第三个标签版本切换即可。但是前提要删除所有非系统数据库。 删除数据库方法: 要在 MongoDB 中删除一个数据库,可以使用 dropDatabase() 命令。请注意,在执行此操作之前&#x…...
16、XSS——会话管理
文章目录 一、web会话管理概述1.1 会话管理1.2 为什么需要会话管理?1.3 常见的web应用会话管理的方式 二、会话管理方式2.1 基于server端的session的管理方式2.2 cookie-based的管理方式2.3 token-based的管理方式 三、安全问题 一、web会话管理概述 1.1 会话管理 …...
稀疏矩阵的操作(数据结构实训)
题目: 标准输入输出 题目描述: 稀疏矩阵可以采用三元组存储。 输入: 输入包含若干个测试用例,每个测试用例的第一行为两个正整数m,n(1<m,n<100),表示矩阵的行数和列数,接下来m行,每行n个整数,表示稀疏…...
sqlite - sqlite3_exec - c++回调函数的处理
文章目录 sqlite - sqlite3_exec - c回调函数的处理概述笔记回调赋值实现用到的数据结构回调分发函数的实现具体的回调处理sqlite3_exe执行完后, 行集的具体处理END sqlite - sqlite3_exec - c回调函数的处理 概述 以前给客户写了个小程序, 处理sqlite执行sql时, 给定回调, 等…...
docker搭建logstash和使用方法
配置logstash 查询下载镜像【固定和elasticsearch一样的版本】 [roothao ~]# docker search logstash NAME DESCRIPTION STARS OFFICIAL AUTOMATED logstash …...
Memory-augmented Deep Autoencoder for Unsupervised Anomaly Detection 论文阅读
Memorizing Normality to Detect Anomaly: Memory-augmented Deep Autoencoder for Unsupervised Anomaly Detection 摘要1.介绍2.相关工作异常检测Memory networks 3. Memory-augmented Autoencoder3.1概述3.2. Encoder and Decoder3.3. Memory Module with Attention-based S…...
Mac端 DevEco Preview 窗口无法展示,提示文件中的node.dir错误
语雀知识库地址:语雀HarmonyOS知识库 飞书知识库地址:飞书HarmonyOS知识库 DevEco版本:Build Version: 3.1.0.501, built on June 20, 2023 环境信息 问题描述 打开 Preview 标签窗口后,提示Preview failed。 Run窗口提示如下 F…...
TIMO后台管理系统 Shiro 反序列化漏洞复现
0x01 产品简介 TIMO 后台管理系统,基于SpringBoot2.0 + Spring Data Jpa + Thymeleaf + Shiro 开发的后台管理系统,采用分模块的方式便于开发和维护,支持前后台模块分别部署,目前支持的功能有:权限管理、部门管理、字典管理、日志记录、文件上传、代码生成等,为快速开发后…...
3.4_1 java自制小工具 - pdf批量转图片
相关链接 目录参考文章:pdf转图片(apache pdfbox)参考文章:GUI界面-awt参考文章:jar包转exe(exe4j)参考文章:IDEA导入GIT项目参考文章:IDEA中使用Gitee管理代码gitee项目链接:pdf_2_image网盘地址…...
vue中实现数字+英文字母组合键盘
完整代码 <template><div class"login"><div click"setFileClick">欢迎使用员工自助终端</div><el-dialog title"初始化设置文件打印消耗品配置密码" :visible.sync"dialogSetFile" width"600px&quo…...
Centos服务器上根据端口号查询jar包,根据jar包查端口号
在开发springboot服务器时,经常会遇到其他人部署的java服务,需要自己维护,留下的信息又非常少。经常面临找不到jar包位置,或者不知道占用端口,不知道启动命令的问题。这里记录一下常用的centos服务器上的命令ÿ…...
数据仓库与数据挖掘复习资料
一、题型与考点[第一种] 1、解释基本概念(中英互译解释简单的含义); 2、简答题(每个10分有两个一定要记住): ① 考时间序列Time series(第六章)的基本概念含义解释作用(序列模式挖掘的作用); ② 考聚类(第五章)重点考…...
限流算法,基于go的gRPC 实现的
目录 一、单机限流 1、令牌桶算法 3、固定窗口限流算法 4、滑动窗口 二、集群限流 1、分布式固定窗口 (基于redis) 2、分布式滑动窗口 一、单机限流 1、令牌桶算法 令牌桶算法是当流量进入系统前需要获取令牌,没有令牌那么就要进行限…...
Shell中HTTP变量和文本处理
在Shell中,HTTP变量和文本处理是常见的任务之一。Shell是一个命令行解释器,可以用来自动化执行各种系统任务。在Shell中,我们可以使用各种命令和工具来处理HTTP变量和文本。 首先,让我们来看看如何在Shell中处理HTTP变量。HTTP变…...
java学习part39map
159-集合框架-Map不同实现类的对比与HashMap中元素的特点_哔哩哔哩_bilibili 1.Map 2.Entry 个人理解是c的pair,代表一个键值对。Map就是entry的叠加 3.常用方法 4.TreeMap 5.Properties...
使用sqoop操作HDFS与MySQL之间的数据互传
一,数据从HDFS中导出至MySQL中 1)开启Hadoop、mysql进程 start-all.sh/etc/init.d/mysqld start/etc/init.d/mysqld status 2)将学生数据stu_data.csv传到HDFS的/local_student目录下 在hdfs中创建目录 hdfs dfs -mkdir /local_student 上…...
Kafka使用指南
Kafka简介架构设计Kafka的架构设计关键概念Kafka的架构设计关键机制 Partition介绍Partition工作机制 应用场景ACK机制介绍ACK机制原理ACK机制对性能的影响ACK控制粒度Kafka分区数对集群性能影响调整分区优化集群性能拓展Kafka数据全局有序 Kafka简介 Kafka是由Apache软件基金…...
HarmonyOS4.0从零开始的开发教程03初识ArkTS开发语言(中)
HarmonyOS(二)初识ArkTS开发语言(中)之TypeScript入门 浅析ArkTS的起源和演进 1 引言 Mozilla创造了JS,Microsoft创建了TS,Huawei进一步推出了ArkTS。 从最初的基础的逻辑交互能力,到具备类…...
西工大计算机学院计算机系统基础实验一(函数编写1~10)
还是那句话,千万不要慌,千万不要着急,耐下性子慢慢来,一步一个脚印,把基础打的牢牢的,一样不比那些人差。回到实验本身,自从按照西工大计算机学院计算机系统基础实验一(…...
VMware 虚拟机 电脑重启后 NAT 模式连不上网络问题修复
问题描述: 昨天 VMware 安装centos7虚拟机,网络模式配置的是NAT模式,配置好后,当时能连上外网,今天电脑重启后,发现连不上外网了 检查下各个配置,都没变动,突然就连不上了 网上查了…...
【桑基图】绘制桑基图
绘制桑基图 一、绘制桑基图(1)方法一:去在线网站直接绘制(2)方法二:写html之后在vscode上运行 二、遇到的问题(1)当导入一些excel的时候,无法绘制出桑基图 一、绘制桑基图…...
ACM32F403/F433 12 位多通道,支持 MPU 存储保护功能,应用于工业控制,智能家居等产品中
ACM32F403/F433 芯片的内核基于 ARMv8-M 架构,支持 Cortex-M33 和 Cortex-M4F 指令集。芯片内核 支持一整套DSP指令用于数字信号处理,支持单精度FPU处理浮点数据,同时还支持Memory Protection Unit (MPU)用于提升应用的…...
7. 从零用Rust编写正反向代理, HTTP及TCP内网穿透原理及运行篇
wmproxy wmproxy是由Rust编写,已实现http/https代理,socks5代理, 反向代理,静态文件服务器,内网穿透,配置热更新等, 后续将实现websocket代理等,同时会将实现过程分享出来ÿ…...
UE4.27-UE5.1设置打包Android环境
打包Android配置文件 1. 配置打包Android的SDK需求文件位于下面文件中: 2. 指定了对应的SDK环境变量名字以及NDK需求等: UE4.27-UE5.1--脚本自动配置 安装前提 1. 务必关闭虚幻编辑器和Epic Games Launcher,以确保NDK组件的安装或引擎环境…...
MySQL授权密码
mysql> crate databases school charcter set utf8; Query OK, 1 row affected, 1 warning (0.00 sec) 2.在school数据库中创建Student和Score表 mysql> use school Database changed mysql> create table student-> -> (id int(10) primary key auto_incremen…...
0X05
打开题目 点击完登录和注册都没有什么反应,所以先扫一下看看 在出现admin.php后就截止了,访问看看,进入后台。。 尝试一下弱口令 admin/12345 或者是demo/demo 设计中-自定义->右上角导出主题 找到一个导出的点,下载了一个1.zip压缩包…...
Doris优化总结
1 查看QueryProfile 利用查询执行的统计结果,可以更好的帮助我们了解Doris的执行情况,并有针对性的进行相应Debug与调优工作。 FE将查询计划拆分成为Fragment下发到BE进行任务执行。BE在执行Fragment时记录了运行状态时的统计值,并将Fragment执行的统计信息输出到日志之中。…...
案例059:基于微信小程序的在线投稿系统
文末获取源码 开发语言:Java 框架:SSM JDK版本:JDK1.8 数据库:mysql 5.7 开发软件:eclipse/myeclipse/idea Maven包:Maven3.5.4 小程序框架:uniapp 小程序开发软件:HBuilder X 小程序…...
利用STM32内置Bootloader实现USB DFU固件升级
本文将介绍如何利用STM32内置的Bootloader来实现USB DFU(Device Firmware Upgrade)固件升级功能。首先,我们会介绍USB DFU的原理和工作流程。然后,我们将详细讲解如何配置STM32芯片以支持USB DFU,并提供相应的代码示例…...
Centos7如何安装MySQL
目录 一、卸载mysql 二、安装mysql 注:本文主要是看了这位大佬安装MySQL,才想着写一篇记录一下。 一、卸载mysql 安装mysql之前一定要将之前安装的mysql相关文件删除干净,防止出现错误。 (1)关闭mysql 开启了mysql就…...
VR远程带看,助力线下门店线上化转型“自救”
VR远程带看,因自身高效的沉浸式在线沟通功能,逐渐走进了大众的视野。身临其境的线上漫游体验以及实时同屏互联的新型交互模式,提升了商家同用户之间的沟通效率,进一步实现了远程线上一对一、一对多的同屏带看,用户足不…...
算法通关村第十七关-白银挑战贪心算法高频题目
大家好我是苏麟 , 今天说说贪心算法的高频题目 . 大纲 区间问题判断区间是否重叠合并区间插入区间 区间问题 判断区间是否重叠 描述 : 给定一个会议时间安排的数组 intervals ,每个会议时间都会包括开始和结束的时间intervalsl[i] [start, end] ,请你…...
【数据结构】动态规划(Dynamic Programming)
一.动态规划(DP)的定义: 求解决策过程(decision process)最优化的数学方法。 将多阶段决策过程转化为一系列单阶段问题,利用各阶段之间的关系,逐个求解。 二.动态规划的基本思想: …...