上篇文章说了,触发rebalance是当消费者组订阅的topic数量发生改变,或者topic分区数量发生改变,或者consumer数量发生变化,比如新的consumer加入组,则会重平衡。还介绍了分区策略range,round-robin,sticky。Kafka监听。以及kafkaConsumer是线程安全的吗?
Broker是kafka非常重要的主键,负责持久化producer端发送的消息,同时还为consumer端提供消息消费。
Broker是一个服务载体,首先我们要深入了解broker内部设计。
一.消息设计
消息引擎,定义消息格式肯定首当其冲,使用什么数据结构来保存消息和 消息队列是第一个要解决的问题。
首先如果用java 类定义消息格式,则必定会受到java对象开销所累。在java内存模型中,java memory model,jmm中保存对象开销其实很大,可能花费比消息大小大两倍的空气来保存数据。为了降低这种开销,jvm通常会对用户自定义类进行重排序,试图减少内存使用,但随着java堆数据越来越多,垃圾回收性能下降的很快,从而整体上拖累吞吐量。
而byteBuffer是紧凑的二进制字节结构,根据kafka官网测试,一台32g的机器上,kafka几乎可用用不到28到30的物理内存而不用担心java的gc糟糕性能。同一条消息比java至少节省百分之40的空间,且扩展性更好。
Kafka消息格式主要有三个版本变迁,V0版本,V1版本和V2版本。
v0版本
主要说的是kafka0.10.0.0之前的版本,是kafka最早的版本。
CRC效验码:4个字节CRC效验码,用于确保消息在传输过程中不会被恶意篡改。
Magic:单字节的版本号。V0版本这个是0,V1版本这个是1,V2版本这个是2.
Attribute:单字节属性字段,只用于后三位来表示消息的积压类型。
Key长度字段:4字节的消息key长度信息,若未指定key,则是-1。
Key值:消息key由key的长度指定。
Value长度字段:4字节的消息value长度信息,若未指定value,则是-1。
Value值:消息value,由value长度指定。
上面除了key和value的所有字段都称为消息头部(message header),总共占有14个字节。
也就是说一条kafka最少占有14个字节。
上面attribute字段占一个字节,目前v0只用低三位来指定压缩类型,其他五位用于扩展。
0x00:未启动压缩。
0x01:GZIP。
0x02:Snappy。
0x03:LZ4。
所以如果真的存值,压缩是lz4,则应该是:
CRC:对所有字段进行crc效验之后的crc值。
Magic:0
Attribule:0x03
Key长度:3
Key:key
Value长度:5
Value:value
所以这条消息的总字节长度是4+1+1+4+3+4+5=22个字节,如果未指定key,而value还是这个值,则组成的是4+1+1+4+4+5 = 19个字节,少的字节正式key值,这样也会花4个字节来存储-1.
v1版本
随着kafka演进,大家渐渐发现:
1. 由于没有时间信息,kafka删除日志只能靠最近修改时间。
2. 很多流处理的框架需要消息的保存时间以便对消息进行操作。
于是在kafka0.10.0.0中改进了消息格式成v1,加入了时间戳,在头部信息多了8个字节的时间戳。
还有一个区别在与attrubite之前只用了后三位,现在用了第四位,用于指定时间戳,当前支持两种时间戳create_time和log_append_time。前者表示消息创建时候由producer指定时间戳,后者表示消息发送到broker端时由broker指定时间戳。
V2版本
这里有个kafka消息集合 和 kafka层次的概念。Kafka无论哪个版本,消息层次都分为两层:消息集合 和 消息。
一个消息集合包含若干个日志项,而每个日志项都封装这实际消息和元数据信息,kafka日志文件就是由一系列消息集合日志构成的。Kafka不会在消息层面直接操作,它总是在消息集合上写入操作。
V0和v1版本更多的使用日志项log entry,而v2版本使用消息批次record batch。
每条消息集合中的日志项由一条“浅层”消息和日志项头部组成。
浅层消息(shallow message):如果没有启动消息压缩,则这条浅层消息就是消息本身。否则,kafka会将多条消息压缩到一起封装进这条浅层消息的value字段。此时浅层消息被称为包装消息(或者外部消息,warpper消息),而value则称为内部消息(inner消息)。V0和v1只包含一条浅层消息。
日志项头部(log entry header):头部由8字节位移(offset)字段加4字节的长度(size)字段构成。注意这里的offset指kafka分区里的offset,并非consumer的offset。如果未压缩,则该offset就是offset本身。否则该字段表示wrapper消息中最后一条inner消息的offset。因此从v0到v1在消息集合日志搜索该日志起始位移是非常困难的,需要遍历kafka所有inner消息。
随着版本的迭代,逐渐发现一些缺点:
1、空间利用率不高:不论key和value长度是多少,总是4个字节保存,例如保存100或者1000都是使用4个字节,但我们起始只要7位就可以保存100,也就是一个字节足够,另外三个字节纯属浪费,这样每天kafka高并发情况下得浪费多少内存。
2、只保存最新消息位移:入上所述,若启用压缩,这个版本中offset是消息集合中最后一条消息的offset,如果用户想获取第一条位移,必须吧所有消息全部解压装入内存,然后反向遍历才可以获取,显然代价比较大。
3、冗余的CRC效验:为每条消息都效验比较鸡肋。鉴于某些情况,对每条消息都效验是浪费cpu内存的。
4、未保存消息长度:每次需要单挑消息的总字节数信息时都要计算,没有使用单独字段来保存。
鉴于这些缺点,kafka0.11.0.0版本重构了消息和消息集合格式的定义,升级成v2版本。
二、集群管理
Kafka是分布式消息引擎集群,它支持自动化服务发现与成员管理。那么他是怎么做到的呢,是依赖zookeeper实现,每当一个broker启动,会将自己注册到zookeeper节点。
首先,每个broker在zookeeper下注册节点的路径是chroot/brokers/ids/<broker.id>。如果没有配置chroot,则路径是/brokers/ids/<broker.id>。是否配置了chroot取决于server.properties中的zookeeper.connect参数是否设置了chroot。
尽管新版本producer发送消息和consumer消费消息不再需要连接zookeeper,但kafka依然依赖zookeeper。甚至某种程度上是kafka单点失效的组件,一旦zookeeper挂掉,kafka很多组件无法使用。
下面介绍下zookeeper的路径:
1、/brokers:里面保存着kafka集群所有消息,包含每台broker注册信息,topic的信息等。
2、/controller:保存着kafka controller组件(controller负责集群的领导者选举)的注册信息,同时也负责controller动态选举。
3、/admin:保存管理脚本的输出结果,比如删除topic,对分区进行重分配等。
4、/isr_change_notification:保存ISR列表发生变化的分区列表。Controller会注册一个监听器时刻监测这下面的变化。
5、/Config:保存kafka集群下的各种资源定制化配置信息。比如每个topic都有专属一组配置,/config/topic/<topic>
6、/Cluster:保存kafka集群简要信息,包括id信息和集群版本号。
7、/controller_epoch:保存了controller组件版本号。Kafka使用该版本号隔离无效的controller请求。