Kafka学习总结（二）——消息存储之FileMessageSet

野性酷女 2022-05-27 12:57 159阅读 0赞

### 参考学习资料：源码[点击打开链接][Link 1] ###

### 参考学习资料：资料[点击打开链接][Link 2] ###

在Kafka中使用FileMessageSet管理日志文件，它对应着磁盘上一个真正的日志文件。FileMessageSet继承了MessaeSet抽象类,MessageSet。保存的数据格式分为三部分：8字节的ofset和4字节的size以及size子集的message 数据，前两个部分被称为LogOverhead。

![Center][]

Kafka使用Message表示消息，Message使用ByteBuffer保存数据，其格式以及各个部分的含义：

![Center 1][]

CRC32： 消息的校验码，4个字节

magic: 魔数标识，与消息格式有关，取值0或者1.magic为0的时候，消息的offset使用绝对offset且消息格式没有timestamp部分；当magic为1的时候，消息的offset使用相对的offset且消息格式存在timestamp部分，所以magic不同消息长度也不一样。

attributes: 消息的属性，1个字节。其中0-2表示消息使用的压缩类型：0表示为无压缩，1表示gzip压缩，2表示snappy压缩

3表示时间戳类型：0表示创建时间，1表示追加时间

timestamp: 时间戳，其含义由attribute的第三位确定

key length: 消息 key的长度

key： 消息的key

value length: 消息value的长度

value: 消息的value

**一 FileMessageSet 核心字段**

file: 指向磁盘上日志文件

channel：FileChannel类型，用于读写对应的日志文件

start：FileMessageSet对象除了表示一个完整的日志文件，还可以表示日志文件的分片，start表示分片的开始位置

end：表示分片的结束位置

isSlice：表示当前FileMessageSet是否为日志文件的分片

\_size：FileMessageSet大小，单位是字节，如果是分片则表示分片大小

### FileMessageSet的功能 ###

*  消息的增删改查
 *  进行必要的检查，比如是否是指定的消息格式（检查Magic值）
 *  进行消息格式的转换

对于最核心的功能——增删改查，我们在这里进一步展开。首先FileMessageSet只处理最外层的消息，而不考虑嵌套的消息，嵌套消息会移交给之前的ByteBufferMessageSet处理。某种程度上，我们也可以把ByteBufferMessageSet看做是嵌套消息。

FileMessageSet的删除也分为两种，一种是从特定位置截断，一种是直接删除整个文件。其查询主要是从消息的序号也就是offset获得其在文件中的位置。其增加只允许向尾部追加，若想在中间添加，必须先截断。

我们列一下几个重要的原子操作吧

*  read(buffer,position,length),read(position,length):FileMessageSet
 *  writeTo(channel,position,size)
 *  truncate(size)
 *  search(offset):position
 *  close
 *  flush

### FileMesssage的设计 ###

FileMessageSet使用FileChannel来进行读写，我们的操作依赖于position进行，需要首先定位。同样，FileMessageSet允许支持切片，也就是截取文件中的一部分，指定start和end。但是这样每次检查末尾都需要考虑end了。

这里首先要注意的第一点是channel的游标应该始终定位在set的尾部，这是为了保证写入是顺序的，所以在初始化的时候就应该将游标移到尾部。

第二点是在关闭channel的时候需要先做flush然后截断。这一点可能不太好理解，这里举个例子，如果我使用了分片，并在位置end后写入了一条新消息，由于必须保证消息是有序的，所以后面所有的消息必须丢弃。这也是保证消息的顺序写特性。

def close() {
            flush()
            trim()
            channel.close()
          }

第三点是迭代的过程，这里面几乎所有的原子操作均是从遍历实现的，遍历中需要进行较多的检查操作，主要是以下几点。

*  如果当前读取的messageSize小于最小的消息头大小，说明消息出现错误
 *  如果当前读取的messageSize大于剩余的容量，说明最后一条消息不完整
 *  如果剩下的容量小于offsetSize+MessageSizeLength，说明已经没有消息了

但是这里的容量需要同时考虑指定的end和channel的结尾，下面以生成迭代器为例。

override def makeNext(): MessageAndOffset = {
                //最后一条消息出现在end之后
                if(location + sizeOffsetLength >= end)
                  return allDone()
        
        
                // read the size of the item
                sizeOffsetBuffer.rewind()
                channel.read(sizeOffsetBuffer, location)
        
                //最后一条消息出现在下一文件中
                if(sizeOffsetBuffer.hasRemaining)
                  return allDone()
        
                sizeOffsetBuffer.rewind()
                val offset = sizeOffsetBuffer.getLong()
                val size = sizeOffsetBuffer.getInt()
        
                //最后一条消息被end截断或消息大小出现问题
                if(size < Message.MinMessageOverhead || location + sizeOffsetLength + size > end)
                  return allDone()
              //消息过大
                if(size > maxMessageSize)
                  throw new CorruptRecordException("Message size exceeds the largest allowable message size (%d).".format(maxMessageSize))
        
                // read the item itself
                val buffer = ByteBuffer.allocate(size)
                channel.read(buffer, location + sizeOffsetLength)
        
                //最后一条消息被文件截断
                if(buffer.hasRemaining)
                  return allDone()
                buffer.rewind()
        
                // increment the location and return the item
                location += size + sizeOffsetLength
                new MessageAndOffset(new Message(buffer), offset)
              }

第四条是追加是以ByteBufferMessageSet为单位的，这主要是将嵌套消息和一般消息还有批量写入统一在一个方法下。

第五条是一个有趣的代码细节

def delete(): Boolean = {
        CoreUtils.swallow(channel.close())
        file.delete()
      }
    
    def swallow(log: (Object, Throwable) => Unit, action: => Unit) {
        try {
          action
        } catch {
          case e: Throwable => log(e.getMessage(), e)
        }
      }

这里将代码块包裹在try catch中，通过这种方法调用的形式，非常简洁优美，有点类似于使用AOP收集异常，值得借鉴。

### 消息读入的过程 ###

写到这儿，让我们来回顾一下整个消息存储的内容并整理出完整的流程吧。

1.  首先FileMessageSet读取最外层消息
2.  若该消息是嵌套消息，则生成ByteBufferMessageSet解压缩并生成原子消息集
3.  通过调用message自身的方法进行检验和获取基本信息比如消息格式
4.  通过MessageAndMeta加上译码器获得key-value对象

### 消息写入的过程 ###

1.  首先MessageWriter写入key-value和消息头生成buffer
2.  对于嵌套消息使用刚刚的buffer生成 ByteBufferMessageSet并convert压缩成新的ByteBufferMessageSet
3.  再使用FileMessageSet追加ByteBUfferMessageSet

[Link 1]: https://www.cnblogs.com/cloud-zhao/p/5748693.html
[Link 2]: https://blog.csdn.net/zhanglh046/article/details/72821427
[Center]: /images/20220527/bee3690eaaf043ecb608222bf2464348.png
[Center 1]: /images/20220527/418e3ce2b09d4e769f59b90ac5d51288.png