一文看懂 Kafka 消息格式的演变

摘要

对于一个成熟的消息中间件而言，消息格式不只关系到功能维度的扩展，还牵涉到性能维度的优化。随着Kafka的迅猛发展，其消息格式也在不断的升级改进，从0.8.x版本开始到如今的1.1.x版本，Kafka的消息格式也经历了3个版本。本文这里主要来说述Kafka的三个版本的消息格式的演变，文章偏长，建议先关注后鉴定。

---

Kafka根据topic（主题）对消息进行分类，发布到Kafka集群的每条消息都须要指定一个topic，每一个topic将被分为多个partition（分区）。每一个partition在存储层面是追加log（日志）文件，任何发布到此partition的消息都会被追加到log文件的尾部，每条消息在文件中的位置称为offset（偏移量），offset为一个long型的数值，它惟一标记一条消息。

每一条消息被发送到Kafka中，其会根据必定的规则选择被存储到哪个partition中。若是规则设置的合理，全部的消息能够均匀分布到不一样的partition里，这样就实现了水平扩展。如上图，每一个partition由其上附着的每一条消息组成，若是消息格式设计的不够精炼，那么其功能和性能都会大打折扣。好比有冗余字段，势必会使得partition没必要要的增大，进而不只使得存储的开销变大、网络传输的开销变大，也会使得Kafka的性能降低；又好比缺乏字段，在最初的Kafka消息版本中没有timestamp字段，对内部而言，其影响了日志保存、切分策略，对外部而言，其影响了消息审计、端到端延迟等功能的扩展，虽然能够在消息体内部添加一个时间戳，可是解析变长的消息体会带来额外的开销，而存储在消息体（参考下图中的value字段）前面能够经过指针偏量获取其值而容易解析，进而减小了开销（能够查看v1版本），虽然相比于没有timestamp字段的开销会差一点。如此分析，仅在一个字段的一增一减之间就有这么多门道，那么Kafka具体是怎么作的呢？本文只针对Kafka 0.8.x版本开始作相应说明，对于以前的版本不作陈述。

v0版本

对于Kafka消息格式的第一个版本，咱们把它称之为v0，在Kafka 0.10.0版本以前都是采用的这个消息格式。注意如无特殊说明，咱们只讨论消息未压缩的情形。

上左图中的“RECORD”部分就是v0版本的消息格式，大多数人会把左图中的总体，即包括offset和message size字段都都当作是消息，由于每一个Record（v0和v1版）一定对应一个offset和message size。每条消息都一个offset用来标志它在partition中的偏移量，这个offset是逻辑值，而非实际物理偏移值，message size表示消息的大小，这二者的一块儿被称之为日志头部（LOG_OVERHEAD），固定为12B。LOG_OVERHEAD和RECORD一块儿用来描述一条消息。与消息对应的还有消息集的概念，消息集中包含一条或者多条消息，消息集不只是存储于磁盘以及在网络上传输（Produce & Fetch）的基本形式，并且是kafka中压缩的基本单元，详细结构参考上右图。

下面来具体陈述一下消息（Record）格式中的各个字段，从crc32开始算起，各个字段的解释以下：

• 1. crc32（4B）：crc32校验值。校验范围为magic至value之间。
• 2. magic（1B）：消息格式版本号，此版本的magic值为0。
• 3. attributes（1B）：消息的属性。总共占1个字节，低3位表示压缩类型：0表示NONE、1表示GZIP、2表示SNAPPY、3表示LZ4（LZ4自Kafka 0.9.x引入），其他位保留。
• 4. key length（4B）：表示消息的key的长度。若是为-1，则表示没有设置key，即key=null。
• 5. key：可选，若是没有key则无此字段。
• 6. value length（4B）：实际消息体的长度。若是为-1，则表示消息为空。
• 7. value：消息体。能够为空，好比tomnstone消息。

v0版本中一个消息的最小长度（RECORD_OVERHEAD_V0）为crc32 + magic + attributes + key length + value length = 4B + 1B + 1B + 4B + 4B =14B，也就是说v0版本中一条消息的最小长度为14B，若是小于这个值，那么这就是一条破损的消息而不被接受。

这里咱们来作一个测试，首先建立一个partition数和副本数都为1的topic，名称为“msg_format_v0”，而后往msg_format_v0中发送一条key=”key”，value=”value”的消息，以后查看对应的日志：

[root@node1 kafka_2.10-0.8.2.1]# bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/msg_format_v0-0/00000000000000000000.log
Dumping /tmp/kafka-logs-08/msg_format_v0-0/00000000000000000000.log
Starting offset: 0
offset: 0 position: 0 isvalid: true payloadsize: 5 magic: 0 compresscodec: NoCompressionCodec crc: 592888119 keysize: 3
复制代码

查看消息的大小，即00000000000000000000.log文件的大小为34B，其值正好等于LOG_OVERHEAD+RECORD_OVERHEAD_V0 + 3B的key + 5B的value = 12B + 14B + 3B + 5B = 34B。

[root@node1 msg_format_v0-0]# ll *.log
-rw-r--r-- 1 root root       34 Apr 26 02:52 00000000000000000000.log
复制代码

咱们再发送一条key=null, value=”value”的消息，以后查看日志的大小：

[root@node3 msg_format_v0-0]# ll *.log
-rw-r--r-- 1 root root       65 Apr 26 02:56 00000000000000000000.log
复制代码

日志大小为65B，减去上一条34B的消息，能够得知本条消息的大小为31B，正好等于LOG_OVERHEAD+RECORD_OVERHEAD_V0 + 5B的value = 12B + 14B+ 5B = 31B。

v1版本

kafka从0.10.0版本开始到0.11.0版本以前所使用的消息格式版本为v1，其比v0版本就多了一个timestamp字段，表示消息的时间戳。v1版本的消息结构图以下所示：

v1版本的magic字段值为1。v1版本的attributes字段中的低3位和v0版本的同样，仍是表示压缩类型，而第4个bit也被利用了起来：0表示timestamp类型为CreateTime，而1表示tImestamp类型为LogAppendTime，其余位保留。v1版本的最小消息（RECORD_OVERHEAD_V1）大小要比v0版本的要大8个字节，即22B。若是像v0版本介绍的同样发送一条key=”key”，value=”value”的消息，那么此条消息在v1版本中会占用42B，具体测试步骤参考v0版的相关介绍。

消息压缩

常见的压缩算法是数据量越大压缩效果越好，一条消息一般不会太大，这就致使压缩效果并不太好。而kafka实现的压缩方式是将多条消息一块儿进行压缩，这样能够保证较好的压缩效果。并且在通常状况下，生产者发送的压缩数据在kafka broker中也是保持压缩状态进行存储，消费者从服务端获取也是压缩的消息，消费者在处理消息以前才会解压消息，这样保持了端到端的压缩。

压缩率是压缩后的大小与压缩前的对比。例如：把100MB的文件压缩后是90MB，压缩率为90/100*100%=90%，压缩率通常是越小压缩效果越好。通常口语化陈述时会误描述为压缩率越高越好，为了不混淆，本文不引入学术上的压缩率而引入压缩效果，这样容易达成共识。

讲解到这里都是针对消息未压缩的状况，而当消息压缩时是将整个消息集进行压缩而做为内层消息（inner message），内层消息总体做为外层（wrapper message）的value，其结构图以下所示：

压缩后的外层消息（wrapper message）中的key为null，因此图右部分没有画出key这一部分。当生产者建立压缩消息的时候，对内部压缩消息设置的offset是从0开始为每一个内部消息分配offset，详细能够参考下图右部：

其实每一个从生产者发出的消息集中的消息offset都是从0开始的，固然这个offset不能直接存储在日志文件中，对offset进行转换时在服务端进行的，客户端不须要作这个工做。外层消息保存了内层消息中最后一条消息的绝对位移（absolute offset），绝对位移是指相对于整个partition而言的。参考上图，对于未压缩的情形，图右内层消息最后一条的offset理应是1030，可是被压缩以后就变成了5，而这个1030被赋予给了外层的offset。当消费者消费这个消息集的时候，首先解压缩整个消息集，而后找到内层消息中最后一条消息的inner offset，而后根据以下公式找到内层消息中最后一条消息前面的消息的absolute offset（RO表示Relative Offset，IO表示Inner Offset，而AO表示Absolute Offset）：

RO = IO_of_a_message - IO_of_the_last_message
AO = AO_Of_Last_Inner_Message + RO
复制代码

注意这里RO是前面的消息相对于最后一条消息的IO而言的，因此其值小于等于0，0表示最后一条消息自身。

压缩消息，英文是compress message，Kafka中还有一个compact message，经常也会被人们直译成压缩消息，须要注意二者的区别。compact message是针对日志清理策略而言的（cleanup.policy=compact），是指日志压缩（log compaction）后的消息，这个后续的系列文章中会有介绍。本文中的压缩消息单指compress message，即采用GZIP、LZ4等压缩工具压缩的消息。

在讲述v1版本的消息时，咱们了解到v1版本比v0版的消息多了个timestamp的字段。对于压缩的情形，外层消息的timestamp设置为：

1. 若是timestamp类型是CreateTime，那么设置的是内层消息中最大的时间戳（the max timestampof inner messages if CreateTime is used）。
2. 若是timestamp类型是LogAppendTime，那么设置的是kafka服务器当前的时间戳；

内层消息的timestamp设置为：

1. 若是外层消息的timestamp类型是CreateTime，那么设置的是生产者建立消息时的时间戳。
2. 若是外层消息的timestamp类型是LogAppendTime，那么全部的内层消息的时间戳都将被忽略。

对于attributes字段而言，它的timestamp位只在外层消息（wrapper message）中设置，内层消息（inner message）中的timestamp类型一直都是CreateTime。

v2版本

kafka从0.11.0版本开始所使用的消息格式版本为v2，这个版本的消息相比于v0和v1的版本而言改动很大，同时还参考了Protocol Buffer而引入了变长整型（Varints）和ZigZag编码。Varints是使用一个或多个字节来序列化整数的一种方法，数值越小，其所占用的字节数就越少。ZigZag编码以一种锯齿形（zig-zags）的方式来回穿梭于正负整数之间，以使得带符号整数映射为无符号整数，这样可使得绝对值较小的负数仍然享有较小的Varints编码值，好比-1编码为1,1编码为2，-2编码为3。详细能够参考：developers.google.com/protocol-bu…。

回顾一下kafka v0和v1版本的消息格式，若是消息自己没有key，那么key length字段为-1，int类型的须要4个字节来保存，而若是采用Varints来编码则只须要一个字节。根据Varints的规则能够推导出0-63之间的数字占1个字节，64-8191之间的数字占2个字节，8192-1048575之间的数字占3个字节。而kafka broker的配置message.max.bytes的默认大小为1000012（Varints编码占3个字节），若是消息格式中与长度有关的字段采用Varints的编码的话，绝大多数状况下都会节省空间，而v2版本的消息格式也正是这样作的。不过须要注意的是Varints并不是一直会省空间，一个int32最长会占用5个字节（大于默认的4字节），一个int64最长会占用10字节（大于默认的8字节）。

v2版本中消息集谓之为Record Batch，而不是先前的Message Set了，其内部也包含了一条或者多条消息，消息的格式参见下图中部和右部。在消息压缩的情形下，Record Batch Header部分（参见下图左部，从first offset到records count字段）是不被压缩的，而被压缩的是records字段中的全部内容。

先来说述一下消息格式Record的关键字段，能够看到内部字段大量采用了Varints，这样Kafka能够根据具体的值来肯定须要几个字节来保存。v2版本的消息格式去掉了crc字段，另外增长了length（消息总长度）、timestamp delta（时间戳增量）、offset delta（位移增量）和headers信息，而且attributes被弃用了，笔者对此作以下分析（对于key、key length、value、value length字段和v0以及v1版本的同样，这里再也不赘述）：

1. length：消息总长度。 2. attributes：弃用，可是仍是在消息格式中占据1B的大小，以备将来的格式扩展。 3. timestamp delta：时间戳增量。一般一个timestamp须要占用8个字节，若是像这里保存与RecordBatch的其实时间戳的差值的话能够进一步的节省占用的字节数。 4. offset delta：位移增量。保存与RecordBatch起始位移的差值，能够节省占用的字节数。 5. headers：这个字段用来支持应用级别的扩展，而不须要像v0和v1版本同样不得不将一些应用级别的属性值嵌入在消息体里面。Header的格式如上图最有，包含key和value，一个Record里面能够包含0至多个Header。

若是对于v1版本的消息，若是用户指定的timestamp类型是LogAppendTime而不是CreateTime，那么消息从发送端（Producer）进入broker端以后timestamp字段会被更新，那么此时消息的crc值将会被从新计算，而此值在Producer端已经被计算过一次；再者，broker端在进行消息格式转换时（好比v1版转成v0版的消息格式）也会从新计算crc的值。在这些相似的状况下，消息从发送端到消费端（Consumer）之间流动时，crc的值是变更的，须要计算两次crc的值，因此这个字段的设计在v0和v1版本中显得比较鸡肋。在v2版本中将crc的字段从Record中转移到了RecordBatch中。

v2版本对于消息集（RecordBatch）作了完全的修改，参考上图左部，除了刚刚说起的crc字段，还多了以下字段：

1. first offset：表示当前RecordBatch的起始位移。 2. length：计算partition leader epoch到headers之间的长度。 3. partition leader epoch：用来确保数据可靠性。 4. magic：消息格式的版本号，对于v2版本而言，magic等于2。 5. attributes：消息属性，注意这里占用了两个字节。低3位表示压缩格式，能够参考v0和v1；第4位表示时间戳类型；第5位表示此RecordBatch是否处于事务中，0表示非事务，1表示事务。第6位表示是不是Control消息，0表示非Control消息，而1表示是Control消息，Control消息用来支持事务功能。 6. last offset delta：RecordBatch中最后一个Record的offset与first offset的差值。主要被broker用来确认RecordBatch中Records的组装正确性。 7. first timestamp：RecordBatch中第一条Record的时间戳。 8. max timestamp：RecordBatch中最大的时间戳，通常状况下是指最后一个Record的时间戳，和last offset delta的做用同样，用来确保消息组装的正确性。 9. producer id：用来支持幂等性。 10. producer epoch：和producer id同样，用来支持幂等性。 11. first sequence：和producer id、producer epoch同样，用来支持幂等性。 12. records count：RecordBatch中Record的个数。

这里咱们再来作一个测试，在1.0.0的kafka中建立一个partition数和副本数都为1的topic，名称为“msg_format_v2”。而后一样插入一条key=”key”，value=”value”的消息，查看日志结果以下：

[root@node1 kafka_2.12-1.0.0]# bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/msg_format_v2-0/00000000000000000000.log --print-data-log
Dumping /tmp/kafka-logs/msg_format_v2-0/00000000000000000000.log
Starting offset: 0
baseOffset: 0 lastOffset: 0 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 0 CreateTime: 1524709879130 isvalid: true size: 76 magic: 2 compresscodec: NONE crc: 2857248333
复制代码

能够看到size字段为76，咱们根据上图中的v2版本的日志格式来验证一下，Record Batch Header部分共61B。Record部分中attributes占1B；timestamp delta值为0，占1B；offset delta值为0，占1B；key length值为3，占1B，key占3B；value length值为5，占1B，value占5B；headers count值为0，占1B, 无headers。Record部分的总长度=1B+1B+1B+1B+3B+1B+5B+1B=14B，因此Record的length字段值为14，编码为变长整型占1B。最后推到出这条消息的占用字节数=61B+14B+1B=76B，符合测试结果。一样再发一条key=null，value=”value”的消息的话，能够计算出这条消息占73B。

这么看上去好像v2版本的消息比以前版本的消息占用空间要大不少，的确对于单条消息而言是这样的，若是咱们连续往msg_format_v2中再发送10条value长度为6,key为null的消息，能够获得：

baseOffset: 2 lastOffset: 11 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false position: 149 CreateTime: 1524712213771 isvalid: true size: 191 magic: 2 compresscodec: NONE crc: 820363253
复制代码

原本应该占用740B大小的空间，实际上只占用了191B，若是在v0版本中这10条消息则须要占用320B的空间，v1版本则须要占用400B的空间，这样看来v2版本又节省了不少的空间，由于其将多个消息（Record）打包存放到单个RecordBatch中，又经过Varints编码极大的节省了空间。

就以v1和v2版本对比而立，至于哪一个消息格式占用空间大是不肯定的，要根据具体状况具体分析。好比每条消息的大小为16KB，那么一个消息集中只能包含有一条消息（参数batch.size默认大小为16384），因此v1版本的消息集大小为12B+22B+16384B=16418B。而对于v2版本而言，其消息集大小为61B+11B+16384B=17086B（length值为16384+，占用3B，value length值为16384，占用大小为3B，其他数值型的字段均可以只占用1B的空间）。能够看到v1版本又会比v2版本节省些许空间。

其实能够思考一下：当消息体越小，v2版本中的Record字段的占用会比v1版本的LogHeader+Record占用越小，以致于某个临界点能够彻底忽略到v2版本中Record Batch Header的61B大小的影响。就算消息体很大，v2版本的空间占用也不会比v1版本的空间占用大太多，几十个字节内，反观对于这种大消息体的大小而言，这几十个字节的大小从某种程度上又能够忽略。

因而可知，v2版本的消息不只提供了相似事务、幂等等更多的功能，还对空间占用提供了足够的优化，整体提高很大。也由此体现一个优秀的设计是多么的重要，虽说咱们不要过分的设计和优化，那么是否能够着眼于前来思考一下？kafka为咱们作了一个很好的榜样。