Broker消息设计--Kafka从入门到精通（十三）

上篇文章说了，触发rebalance是当消费者组订阅的topic数量发生改变，或者topic分区数量发生改变，或者consumer数量发生变化，比如新的consumer加入组，则会重平衡。还介绍了分区策略range，round-robin，sticky。Kafka监听。以及kafkaConsumer是线程安全的吗？

Rebalance&多线程实例消费（十二）

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请求。