记录下和kafka相关的Message、日志文件、索引文件、consumer记录消费的offset相关内容,文中很多理解参考文末博文、书籍还有前辈。
kafka中的消息
kafka中的消息Message,在V1版本中是如下部分组成,主要关系key和value。
(1)key:当需要将消息写入到某个topic下的指定partition分区时,需要给定key的值。
(2)value:实际消息内容保存在这里。
(3)其他均是消息的元数据,一般不用关心,对用户来说是透明的。
为了保存这些消息数据,kafka使用了ByteBuffer来存储,它是紧凑型字节数组,相比使用java对象来保存消息数据到堆内存,它更加的节省空间,提高内存使用率。
log和index文件
基本介绍
查看一个topic分区目录下的内容,发现有log、index和timeindex三个文件,它有以下几个特点。
(1)log文件名是以文件中第一条message的offset来命名的,实际offset长度是64位,但是这里只使用了20位,应付生产是足够的。可以看出第一个log文件名是以0开头,而第二个log文件是4161281,说明第一log文件保存了offset从0到4161280的消息。
(2)一组index+log+timeindex文件的名字是一样的,并且log文件默认写满1G后,会进行log rolling形成一个新的组合来记录消息,这个是通过broker端log.segment.bytes=1073741824指定的,可以修改这个值进行调整。
(3)index和timeindex在刚使用时会分配10M的大小,当进行log rolling后,它会修剪为实际的大小,所以看到前几个索引文件的大小,只有几百K。
# 一个分区目录下文件内容,参考文末书籍杜撰,主要为了说明概念
[[email protected] /home/software/kafka-2/kafka-logs/football-0]# ll -h
-rw-r--r--. 1 root root 514K Mar 20 16:04 00000000000000000000.index
-rw-r--r--. 1 root root 1.0G Mar 17 03:36 00000000000000000000.log
-rw-r--r--. 1 root root 240K Mar 20 16:04 00000000000000000000.timeindex
-rw-r--r--. 1 root root 512K Mar 20 16:04 00000000000004161281.index
-rw-r--r--. 1 root root 1.0G Mar 17 03:36 00000000000004161281.log
-rw-r--r--. 1 root root 177K Mar 20 16:04 00000000000004161281.timeindex
-rw-r--r--. 1 root root 10M Mar 20 16:04 00000000000008749921.index
-rw-r--r--. 1 root root 390M Mar 17 03:36 00000000000008749921.log
-rw-r--r--. 1 root root 10M Mar 20 16:04 00000000000008749921.timeindex
如果想查看这些文件,可以使用kafka提供的shell来完成,几个关键信息如下:
(1)offset是逐渐增加的整数。
(2)position是相对外层batch的位置增量,可以理解为消息的字节偏移量。
(3)CreateTime:时间戳。
(4)magic:2代表这个消息类型是V2,如果是0则代表是V0类型,1代表V1类型。本机是V2类型的,不过也可以暂时按照上面的V1来参考理解,具体需要看文末书籍里的详细介绍。
(5)compresscodec:None说明没有指定压缩类型,kafka目前提供了4种可选择,0-None、1-GZIP、2-snappy、3-lz4。
(6)crc:对所有字段进行校验后的crc值。
# 查看并打印log文件内容
[[email protected] /home/software/kafka-2/kafka-logs/football-0]# ../../bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files 00000000000000000004.log --print-data-log
Dumping 00000000000000000004.log
Starting offset: 4
baseOffset: 4 lastOffset: 4 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 49 isTransactional: false position: 0 CreateTime: 1584368524633 isvalid: true size: 85 magic: 2 compresscodec: NONE crc: 3049289418
baseOffset: 5 lastOffset: 5 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 49 isTransactional: false position: 85 CreateTime: 1584368668414 isvalid: true size: 73 magic: 2 compresscodec: NONE crc: 2267711305
baseOffset: 6 lastOffset: 6 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 49 isTransactional: false position: 158 CreateTime: 1584368679882 isvalid: true size: 78 magic: 2 compresscodec: NONE crc: 789213838
baseOffset: 7 lastOffset: 7 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 49 isTransactional: false position: 236 CreateTime: 1584368695371 isvalid: true size: 95 magic: 2 compresscodec: NONE crc: 703634716
结构原理
(1)消息内容,保存在log日志文件中,它是记录message的载体。消息会封装成Record的形式,append到log日志文件末尾,采用的是顺序写模式,参考官网图片,一个topic的不同分区,可以想成queue,都会顺序写入发送到它的消息。图中partition0中的0、1、2、3等数字就是一个分区中消息的offset,它是递增的数字。
注意消费者也是有offset的,刚开始学的时候两者混淆了,消费者的offset指的是消费的位置,它是不断更新的数字,主要是为了下次继续消费定位用的。如官网中图片所示,消费者A消费的offset是9,消费者B消费的offset是11,不同的消费者offset是交给它们自己单独记录的。
(2)位置索引,保存在index文件中,log日志默认每写入4K(log.index.interval.bytes设定的),会写入一条索引信息到index文件中,因此索引文件是稀疏索引,它不会为每条日志都建立索引信息。
下图是网上拿来的直接用了,log文件中的日志,是顺序写入的,由message+实际offset+position组成,索引文件的数据结构则是由相对offset(4byte)+position(4byte)组成,由于保存的是相对第一个消息的相对offset,只需要4byte就可以了,可以节省空间,在实际查找后还需要计算回实际的offset,这对用户是透明的。如下图由于log文件名是从0开始的,因此相对offset为3的实际offset是3+0,依然是3。
对于稀疏索引,尽管它的索引密度不高,但是offset是有序的,kafka查找一条offset对应的实际的消息时,可以通过index二分查找,获取到最近的低位offset,然后从低位offset对应的position开始,从实际的log文件中开始往后查找对应的消息。如要查找offset=5的消息,先去索引文件中找到低位的3 4597这条数据,然后通过4597这个字节偏移量,从log文件中从4597个字节开始读取,直到读取到offset=5的这条数据,这比直接从log文件开始读取要节省时间。二分查找的时间复杂度为O(lgN),如果从头遍历时间复杂度是O(N)。
注意下图的index中逗号是不存在的,这个图片加的逗号是为了方便理解。
(3)时间戳索引文件,它的作用是可以让用户查询某个时间段内的消息,它一条数据的结构是时间戳(8byte)+相对offset(4byte),如果要使用这个索引文件,首先需要通过时间范围,找到对应的相对offset,然后再去对应的index文件找到position信息,然后才能遍历log文件,它也是需要使用上面说的index文件的。
但是由于producer生产消息可以指定消息的时间戳,这可能将导致消息的时间戳不一定有先后顺序,因此尽量不要生产消息时指定时间戳。
消费组和coordinator
消费者消费消息时,会记录消费者offset(注意不是分区的offset,不同的上下文环境一定要区分),这个消费者的offset,也是保存在一个特殊的内部分区,叫做__consumer_offsets,它就一个作用,那就是保存消费组里消费者的offset。默认创建时会生成50个分区(offsets.topic.num.partitions设置),一个副本,如果50个分区分布在50台服务器上,将大大缓解消费者提交offset的压力。可以在创建消费者的时候产生这个特殊消费组。
# 如果只启动了hadoop03一个broker,则所有的50个分区都会在这上面生成
[[email protected] /home/software/kafka-2/bin]# sh kafka-console-consumer.sh --bootstrap-server hadoop03:9092 --topic football --from-beginning --new-consumer
那么问题来了,消费者的offset到底保存到哪个分区呢,kafka中是按照消费组group.id来确定的,使用Math.abs(groupId.hashCode())%50,来计算分区号,这样就可以确定一个消费组下的所有的消费者的offset,都会保存到哪个分区了.
那么问题又来了,既然一个消费组内的所有消费者都把offset提交到了__consumer_offsets下的同一个分区,如何区分不同消费者的offset呢?原来提交到这个分区下的消息,key是groupId+topic+分区号,value是消费者offset。这个key里有分区号,注意这个分区号是消费组里消费者消费topic的分区号。由于实际情况下一个topic下的一个分区,只能被一个消费组里的一个消费者消费,这就不担心offset混乱的问题了。
实际上,topic下多个分区均匀分布给一个消费组下的消费者消费,是由coordinator来完成的,它会监听消费者,如果有消费者宕机或添加新的消费者,就会rebalance,使用一定的策略让分区重新分配给消费者。如下图所示,消费组会通过offset保存的位置在哪个broker,就选举它作为这个消费组的coordinator,负责监听各个消费者心跳了解其健康状况,并且将topic对应的leader分区,尽可能平均的分给消费组里的消费者,根据消费者的变动,如新增一个消费者,会触发coordinator进行rebalance。
还有一个细节,消费者组和coordinator之间还进行了什么通信,各个消费者之间是如何做到默契不抢别人的资源?参考前辈整理如下。
(1)消费组会对选出的coordinator发送join group请求。
(2)coordinator会在消费组中选一个leader消费者,并且随后把要消费的topic信息返回给这个leader。
(3)leader消费者会根据topic信息,指定出一套符合自己消费组的消费方案,通过sync group请求返回给coordinator。
(4)coordinator收到分配方案后会分发给各个消费者。
(5)最后每个消费者身上都会有一套消费方案,都遵守它进行消费。
rebalance
rebalance是消费组内达成一致如何消费topic分区的协议,文末书籍里提到有三个触发条件,这里只记录第一个因为它最常出现,那就是消费组里消费者或增加、或离去、或奔溃(它像极了人生)。其他两个,一个是topic分区数使用kafka shell增加了分区,还有一个就是消费的topic是按照正则去匹配的,当有了符合这个规则的新的topic出现,也会触发rebalance。
它有三种策略,为range、round robin、sticky。
假设topicA分区有p0~p6 一共6个分区,某个消费组有三个消费者,以此为基础来直观感受三个策略。
(1)range
有点类似python的range,它就是一个范围,会按照分区号来划分,结果就是:
消费者1 p0 p1,消费者2 p2 p3,消费者3 p4 p5
(2)round robin
就是随机均匀分配,结果略。
(3)sticky
上面两种分配存在一个小问题,就是有消费者宕机后,重新分配后,原本属于一个消费者消费得好好的的分区会被分到新的消费者。如range策略下消费者3挂掉,重新分配后会变成消费者1 p0 p1 p2 消费者2 p3 p4 p5,这样p2就被重分配了。考虑到管理消费者offset的复杂性,尽量希望维持原来的习惯,如果是sticky策略会变成消费者1 p0 p1 p4 消费者2 p2 p3 p5。
以上,理解不一定正确,写的也比较啰嗦,但学习就是一个不断了解和纠错的过程。
参考博文:
(1)https://blog.csdn.net/xiaoyu_bd/article/details/52398265
(2)《Apache Kafka实战》
原文地址:https://www.cnblogs.com/youngchaolin/p/12543436.html