Kafka 日志模块（三）索引

abstract class AbstractIndex(@volatile var file: File, val baseOffset: Long, val maxIndexSize: Int = -1,
                             val writable: Boolean) extends Closeable {

  protected def entrySize: Int
  private[this] var _maxEntries: Int = mmap.limit() / entrySize
  protected var _entries: Int = mmap.position() / entrySize
}                                 

@volatile
  protected var mmap: MappedByteBuffer = {
    //① 创建索引文件  
    val newlyCreated = file.createNewFile()
    //② 以 writable 指定的方式（读写方式或只读方法）打开索引文件
    val raf = if (writable) new RandomAccessFile(file, "rw") else new RandomAccessFile(file, "r")
    try {
      /* pre-allocate the file if necessary */
      if(newlyCreated) {
        // 预设的索引文件大小不能太小，如果连一个索引项都保存不了，直接抛出异常  
        if(maxIndexSize < entrySize)
          throw new IllegalArgumentException("Invalid max index size: " + maxIndexSize)
        // ③ 设置索引文件长度，roundDownToExactMultiple计算的是不超过maxIndexSize的最大整数倍entrySize 比如maxIndexSize=1234567，entrySize=8，那么调整后的文件长度为1234560
        raf.setLength(roundDownToExactMultiple(maxIndexSize, entrySize))
      }

      /* memory-map the file */
      //④ 更新索引长度字段_length
      _length = raf.length()
      //⑤ 创建MappedByteBuffer对象
      val idx = {
        if (writable)
          raf.getChannel.map(FileChannel.MapMode.READ_WRITE, 0, _length)
        else
          raf.getChannel.map(FileChannel.MapMode.READ_ONLY, 0, _length)
      }
      /* set the position in the index for the next entry */
      //⑥ 如果是新创建的索引文件，将MappedByteBuffer对象的当前位置置成0 如果索引文件已存在，将MappedByteBuffer对象的当前位置设置成最后一个索引项所在的位置
      if(newlyCreated)
        idx.position(0)
      else
        // if this is a pre-existing index, assume it is valid and set position to last entry
        idx.position(roundDownToExactMultiple(idx.limit(), entrySize))
        // ⑦ 返回创建的MappedByteBuffer对象
      idx
    } finally {
        // 关闭打开索引文件句柄
      CoreUtils.swallow(raf.close(), AbstractIndex)
    }
  }

private def indexSlotRangeFor(idx: ByteBuffer, target: Long, searchEntity: IndexSearchEntity): (Int, Int) = {
    // check if the index is empty
    // ① 如果索引为空，直接返回<-1,-1> 对
    if(_entries == 0)
      return (-1, -1)

    // 二分查找算法
    def binarySearch(begin: Int, end: Int) : (Int, Int) = {
      // binary search for the entry
      var lo = begin
      var hi = end
      while(lo < hi) {
        val mid = (lo + hi + 1) >>> 1
        val found = parseEntry(idx, mid)
        // 查找的位移值不能小于当前最小位移值
        val compareResult = compareIndexEntry(found, target, searchEntity)
        if(compareResult > 0)
          hi = mid - 1
        else if(compareResult < 0)
          lo = mid
        else
          return (mid, mid)
      }
      (lo, if (lo == _entries - 1) -1 else lo + 1)
    }

    //② 确认热区首个索引项位于哪个槽，_warmEntries 就是所谓的分割线，固定为8
    // 如果是OffsetIndex，_warmEntries = 8192 / 8 = 1024，即第1024个槽 
    // 如果是TimeIndex，_warmEntries = 8192 / 12 = 682，即第682个槽
    val firstHotEntry = Math.max(0, _entries - 1 - _warmEntries)
    // check if the target offset is in the warm section of the index
    //③ 判断 target 位移值在热区还是冷区
    if(compareIndexEntry(parseEntry(idx, firstHotEntry), target, searchEntity) < 0) {
       // 如果在热区，搜索热区 
      return binarySearch(firstHotEntry, _entries - 1)
    }

    // check if the target offset is smaller than the least offset
    //④ 确保 target 位移值不能小于当前最小位移值
    if(compareIndexEntry(parseEntry(idx, 0), target, searchEntity) > 0)
      return (-1, 0)

    //⑤ 如果在冷区，搜索冷区。
    binarySearch(0, firstHotEntry)
  }    

class OffsetIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writable: Boolean = true)
    extends AbstractIndex(_file, baseOffset, maxIndexSize, writable) {
  import OffsetIndex._

  override def entrySize = 8

def append(offset: Long, position: Int): Unit = {
    inLock(lock) {
      //① 判断索引文件是否未写满
      require(!isFull, "Attempt to append to a full index (size = " + _entries + ").")
      //② 必须满足以下条件之一才允许写入索引项
      //1. 当前索引文件为空
      //2. 要写入的位移大于当前所有已写入的索引项的位移 — kafka 规定索引项中的位移必须是单调增加的
      if (_entries == 0 || offset > _lastOffset) {
        trace(s"Adding index entry $offset => $position to ${file.getAbsolutePath}")
        //③ 在 mmap 中写入物理位置信息
        mmap.putInt(relativeOffset(offset))
        mmap.putInt(position)
        //④ 更新其他元数据统计信息，如当前索引项计数器 _entries 和当前索引项最新位移值 _lastOffset
        _entries += 1
        _lastOffset = offset
        //⑤ 执行校验，写入索引项格式必须符合要求，即索引项个数 * 单个索引项占用字节数匹配当前文件物理大小，否则说明文件已损坏
        require(_entries * entrySize == mmap.position(), entries + " entries but file position in index is " + mmap.position() + ".")
      } else {
        // 如果第 ② 步中两个条件都不满足，不能执行写入索引项操作，抛出异常
        throw new InvalidOffsetException(s"Attempt to append an offset ($offset) to position $entries no larger than" +
          s" the last offset appended (${_lastOffset}) to ${file.getAbsolutePath}.")
      }
    }
  }

  def lookup(targetOffset: Long): OffsetPosition = {
    //加锁
    maybeLock(lock) {
      //创建一个副本
      val idx = mmap.duplicate
      //二分查找
      val slot = largestLowerBoundSlotFor(idx, targetOffset, IndexSearchType.KEY)
      //如果找到则调用 parseEntry 方法 返回offset和postition的映射关系
      //如果找不到则返回 baseOffset, 0
      if(slot == -1)
        OffsetPosition(baseOffset, 0)
      else
        parseEntry(idx, slot)
    }
  }

def maybeAppend(timestamp: Long, offset: Long, skipFullCheck: Boolean = false): Unit = {
    inLock(lock) {
      if (!skipFullCheck)
        require(!isFull, "Attempt to append to a full time index (size = " + _entries + ").")
      // We do not throw exception when the offset equals to the offset of last entry. That means we are trying
      // to insert the same time index entry as the last entry.
      // If the timestamp index entry to be inserted is the same as the last entry, we simply ignore the insertion
      // because that could happen in the following two scenarios:
      // 1. A log segment is closed.
      // 2. LogSegment.onBecomeInactiveSegment() is called when an active log segment is rolled.
      if (_entries != 0 && offset < lastEntry.offset)
        throw new InvalidOffsetException(s"Attempt to append an offset ($offset) to slot ${_entries} no larger than" +
          s" the last offset appended (${lastEntry.offset}) to ${file.getAbsolutePath}.")
      if (_entries != 0 && timestamp < lastEntry.timestamp)
        throw new IllegalStateException(s"Attempt to append a timestamp ($timestamp) to slot ${_entries} no larger" +
          s" than the last timestamp appended (${lastEntry.timestamp}) to ${file.getAbsolutePath}.")
      // We only append to the time index when the timestamp is greater than the last inserted timestamp.
      // If all the messages are in message format v0, the timestamp will always be NoTimestamp. In that case, the time
      // index will be empty.
      if (timestamp > lastEntry.timestamp) {
        trace(s"Adding index entry $timestamp => $offset to ${file.getAbsolutePath}.")
        mmap.putLong(timestamp)
        mmap.putInt(relativeOffset(offset))
        _entries += 1
        _lastEntry = TimestampOffset(timestamp, offset)
        require(_entries * entrySize == mmap.position(), _entries + " entries but file position in index is " + mmap.position() + ".")
      }
    }
  }

def lookup(targetTimestamp: Long): TimestampOffset = {
    maybeLock(lock) {
      val idx = mmap.duplicate
      val slot = largestLowerBoundSlotFor(idx, targetTimestamp, IndexSearchType.KEY)
      if (slot == -1)
        TimestampOffset(RecordBatch.NO_TIMESTAMP, baseOffset)
      else
        parseEntry(idx, slot)
    }
  }