源码阅读-Record

Kafka消息格式相关的代码位于clients工程下的/java/org/apache/kafka/common/record包下。

类之间的关系图如下所示：

1. DefaultRecord（默认消息）

类中的消息格式定义（org.apache.kafka.common.record.DefaultRecord）：

字段	说明
sizeInBytes	消息总长度字段
attributes	消息属性字段
offset	位移增量
timestamp	时间戳增量
sequence	用于支持消息的幂等性
key	key值
value	value值
headers	消息头部属性

kafka将record组装成buffer进行输出（org.apache.kafka.common.record.DefaultRecord#writeTo）：

/*** Write the record to `out` and return its size.*/
public static int writeTo(DataOutputStream out,int offsetDelta,long timestampDelta,ByteBuffer key,ByteBuffer value,Header[] headers) throws IOException {// 消息总数int sizeInBytes = sizeOfBodyInBytes(offsetDelta, timestampDelta, key, value, headers);ByteUtils.writeVarint(sizeInBytes, out);// 属性byte attributes = 0; // there are no used record attributes at the momentout.write(attributes);// 时间增量ByteUtils.writeVarlong(timestampDelta, out);// 位移增量ByteUtils.writeVarint(offsetDelta, out);// keyif (key == null) {ByteUtils.writeVarint(-1, out);} else {// key sizeint keySize = key.remaining();ByteUtils.writeVarint(keySize, out);Utils.writeTo(out, key, keySize);}// valueif (value == null) {ByteUtils.writeVarint(-1, out);} else {// value sizeint valueSize = value.remaining();ByteUtils.writeVarint(valueSize, out);Utils.writeTo(out, value, valueSize);}// headersif (headers == null)throw new IllegalArgumentException("Headers cannot be null");// headers个数ByteUtils.writeVarint(headers.length, out);for (Header header : headers) {// header keyString headerKey = header.key();if (headerKey == null)throw new IllegalArgumentException("Invalid null header key found in headers");byte[] utf8Bytes = Utils.utf8(headerKey);// header key lengthByteUtils.writeVarint(utf8Bytes.length, out);out.write(utf8Bytes);// header valuebyte[] headerValue = header.value();if (headerValue == null) {ByteUtils.writeVarint(-1, out);} else {// header value lengthByteUtils.writeVarint(headerValue.length, out);out.write(headerValue);}}return ByteUtils.sizeOfVarint(sizeInBytes) + sizeInBytes;
}

消息格式可以用如下表示：

2. DefaultRecordBatch（默认消息集合）

类中的消息集合格式的定义（org.apache.kafka.common.record.DefaultRecordBatch）：

消息批次格式可以用如下表示：

源码阅读-Record相关推荐

TiDB 源码阅读系列文章（六）Select 语句概览
在先前的 TiDB 源码阅读系列文章(四) 中,我们介绍了 Insert 语句,想必大家已经了解了 TiDB 是如何写入数据,本篇文章介绍一下 Select 语句是如何执行.相比 Insert,Sel ...
代码分析：NASM源码阅读笔记
NASM源码阅读笔记 NASM(Netwide Assembler)的使用文档和代码间的注释相当齐全,这给阅读源码提供了很大的方便.按作者的说法,这是一个模块化的,可重用的x86汇编器, 而且能够被 ...
Soul网关源码阅读（十）自定义简单插件编写
Soul网关源码阅读(十)自定义简单插件编写简介综合前面所分析的插件处理流程相关知识,此次我们来编写自定义的插件:统计请求在插件链中的经历时长编写准备首先我们先探究一下,一个P ...
Android TV TIF源码阅读笔记
Android TV TIF源码阅读笔记 1.SystemSever.java if (mPackageManager.hasSystem ...
TiDB 源码阅读系列文章（六）Select 语句概览 1
在先前的 TiDB 源码阅读系列文章(四) 中,我们介绍了 Insert 语句,想必大家已经了解了 TiDB 是如何写入数据,本篇文章介绍一下 Select 语句是如何执行.相比 Insert,Sel ...
Apollo perception源码阅读 | lidar
#! https://zhuanlan.zhihu.com/p/389572382 Apollo perception源码阅读 | lidar 本文为Apollo感知融合源码阅读笔记,建议参照Apol ...
Zookeeper源码阅读（一）Jute和传输协议
前言最近开始了Zookeeper的源码阅读和分析,也从现在开始把之前和现在学习到的一些Zookeeper的源码知识和我的一些理解放到博客上.不得不说这是自己第一次去完整的看一个开源项目的完整源码,从 ...
应用监控CAT之cat-client源码阅读（一）
CAT 由大众点评开发的,基于 Java 的实时应用监控平台,包括实时应用监控,业务监控.对于及时发现线上问题非常有用.(不知道大家有没有在用) 应用自然是最初级的,用完之后,还想了解下其背后的原理, ...
centos下将vim配置为强大的源码阅读器
每日杂事缠身,让自己在不断得烦扰之后终于有了自己的清静时光来熟悉一下我的工具,每次熟悉源码都需要先在windows端改好,拖到linux端,再编译.出现问题,还得重新回到windows端,这个过程太耗 ...

源码阅读-Record

源码阅读-Record

1. DefaultRecord（默认消息）

2. DefaultRecordBatch（默认消息集合）

源码阅读-Record相关推荐

最新文章

热门文章