学习背景
接着上篇,我们来聊聊kafka数据在服务端怎么写入的
服务端写入
在介绍服务端的写流程之前,我们先要理解服务端的几个角色之间的关系。
假设我们有一个由3个broker组成的kafka集群,我们在这个集群上创建一个topic叫做shitu-topic,他有10个分区,每个分区有3个副本。那么partition和broker的关系假设如下。
因为每个partition有3个副本,所以每个partition的副本都会均匀的分布在这三台机器上,我们取shitu-topic-0的副本来观察。
在三个broker上,每个broker的log存储目录都有一个shitu-topic-0目录,我们可以成为shitu-topic-0分区,但是同一个时间,只有broker-0上的leader副本对外提供服务,broker-1和broker-2需要去到broker-0上同步消息。在shitu-topic-0目录下就是存储的实际的日志文件。日志文件里包含三个主要的文件内容.log文件存储实际的消息,.index文件存储索引,.timeindex文件存储时间索引。我们把这三个文件合称为一个logsegment日志段,每个log文件只要超过1G就会产生一个新的段文件。日志段文件的命名是以当前段内第一条消息的offset来命名的,这里因为是新创建的topic,第一条消息是0,所以都是0。因为消息是顺序写入的,所以只有最后一个日志段是激活的我们称为active segemnt,活跃段。比如这里活跃段就是00000000000020123000开头的段。
研究消息的写入,就是研究这些文件时怎么产生的,让我们来看一下段文件里每个文件的组织格式。
写入文件
log文件
.log文件存储实际的写入日志,也就是实际的数据存储位置。kafka的log文件存储格式经过了3次变化,目前使用的日志格式称为V2版本,我们取这个版本的日志格式来做讲解。
上图左侧显示的是log文件的格式,我们把log文件内存储的的消息集合称为record batch,而每条消息我们称为一条record,每条record的格式如右边所示。record batch内的字段主要记录的整个log文件的全局属性,比如log文件的起始偏移量,文件长度,epoch,时间戳等等,不做详细解释,也不是重点。
我们说一下每条消息的格式,我们知道每条消息除了实际的消息内容value外,伴随着每条消息的产生,还会产生这条消息的额外附带的信息,比如消息的偏移量,offsset,时间戳timestamp等等。kafka在设计消息的存储时花了很大的心思。
这里我解释一下varint,varlong类型,简单的说,就是可变长的类型。比如一条消息的偏移量是int存储容量是4字节,比如存储10这个偏移量,虽然前面大部分是0,但是实际存储还是需要4字节。而varint则可以根据数据的范围选择合适的存储,比如还是10,那么实际记录这个值1个字节就够了。这样,当写入消息时,比如写入2条消息,偏移量分别是10和11,如果分别存储这两个偏移量,需要
2 * 4B = 8b
而如果使用varint存储,则只需要
2 * 1B + 4B(基础偏移量) = 6B
这里如果不是2条消息,而是10000条消息,那么这个优化就会非常有用。kafka这么做是为了尽最大的可能使用存储空间。当然除了数据格式上的优化,kafka还对数据进行了压缩,也就是records是可以配置不同的压缩算法进行压缩的,比如ZIP。
index文件
.index记录偏移量到实际消息的映射关系。一个很简单的述求,我们想知道某个偏移量的日志的内容,那么我们就需要一种能根据偏移量定位到消息的格式。
index文件的格式由相对偏移量realtive offset和物理偏移量position组成。当一条消息写入时,根据消息的偏移量计算出这条消息的相对偏移量,比如写入的是20123025这条消息,那么用20123025-20123000 = 25得到相对偏移量25,再记录下这消息的起始物理地址1024,即可组成对这条消息的索引。需要·注意的是,这里的索引是稀疏索引,也就是不是每条消息都会产生索引,而是每隔一些消息产生索引,这样能减少索引的文件大小。
每一条索引的需要4B的相对偏偏移量和4B的物理地址偏移量,一共8B,kafka的服务端在设置index文件最大大小时要求index文件必须是索引项的整数倍,如果不是,则会自动转换成最接近的整数倍的数字。
大家这里肯定很好奇那么怎么利用相对偏移量来查找消息,我们解释一下,其实对消息的查找可以概述为根据二分法查找。比如想要查找20123050这条偏移量的消息,先根据这个偏移量,去到我们当前副本的segement集合中根据segement的起始偏移量找到对应的segement,所有的segement的信息是根据相对偏移量以跳表的形式记录的。找到的对应的segement后先计算出相对偏移量20123050-20123000 = 50,然后根据50这个相对偏移量,我们去到相对偏量数组里,使用二分查找找到[20,75]这个相对偏移量范围,那么我们可以在log文件里从1024字节开始,逐条消息的解析,并计算出消息的偏移量是不是50,直到2147字节这个结束的位置为止。如果能找到,说明消息在本partition内,不能我们再换另外的partition查找。
timeindex文件
timeidnex记录时间戳到实际消息的映射关系,我们介绍了index文件的格式,再来理解timeindex文件的格式就容易多了。timeindex文件和index文件的格式类似,由时间戳相对偏移量和消息相对偏移量组成。时间戳相对偏移量根据消息的写入时间来计算,比如写入时间是1733001000,用这个写入的时间减去timeindex文件的起始时间1733000000得到1000这个相对时间戳偏移量。
timeindex文件的查找我们就不说了,大家可以参考index文件。需要注意一点timeindex文件的时间戳是可以设置的,虽然一般kafka服务端会采取自动设置消息写入时间的配置,即log.message.timestamp.type=LogAppendTime,这种情况下因为时间戳由服务器端设置,能够保证时间戳递增。但是如果服务端设置的是CreateTime,并且producer自己设置了消息的生产时间,那么有可能造成timeIndex的写入失败,因为timeindex要求写入的时间必须是递增的。如果不递增,则拒绝本次写入。还有就是,timeindex文件和index文件虽然都是索引,但是他们并不是每条索引项一一对应的,大家从图中也能看出来。
根据timeindex查找对应消息的过程也和index文件的查找类似,不过因为timeindex本身是根据时间戳来查找,所以会有一步先查找每个timeindex文件的最大时间戳,直到找到一个大于查找时间并且最接近查找时间的timeindex文件。这里有点绕,举个例子,第一个timeindex文件的最大时间戳10000,第二个timeindex文件最大时间戳23000,第三个timeindex文件最大时间戳50000,要查找时间戳为15000的消息,那么因为timeindex文件的时间戳是顺序递增的,很明显,第三个文件的消息都是在15000之后产生的,第一个文件的消息都是在15000之前产生的,那么理所应当的,正好拥有大于15000的时间戳23000的第二个文件理论上应该包含15000这个时间戳写入的消息,所以找到第二个文件。找到对应的文件后再去到到对应的这个timeindex文件根据时间偏移量索引找到这个对应的消息(找不到就换partition)。
写入过程
介绍完毕实际的文件内容,我们再来归纳一下数据的写入过程。这里不会介绍副本之间的同步的问题,只介绍在leader副本上数据的写入。
当消息通过client发送到broker上时,broker根据消息的topic找到这个topic的leadder副本。leadter副本根据消息的信息计算出消息归属的parititon。找到parititon后根据偏移量设置计算出消息的偏移量和时间戳,再找到对应的active segement,在index文件中追加消息,并根据需要决定是否写入index文件和timeindex文件。