kafka源码阅读（4）-索引查找

if (target > indexEntry[end - N]) // if the target is in the last N entries of the index
	binarySearch(end - N, end)
else
	binarySearch(begin, end - N)

abstract class AbstractIndex(
  // 索引文件
  @volatile var file: File,
  // 索引基础偏移量，文件以该偏移量命名，比如00000000123.index
  val baseOffset: Long,
  // 索引文件最大字节大小，由broker端的segment.index.bytes参数确定，默认是10MB
  val maxIndexSize: Int = -1,
  // 索引文件打开方式，true表示文件可读写，false表示只读
  val writable: Boolean
) extends Closeable {
  // 索引项大小，由具体子类定义
  protected def entrySize: Int
  // 即上一小节中的N
  protected def _warmEntries: Int = 8192 / entrySize
}

@volatile
protected var mmap: MappedByteBuffer = {
  // 创建索引文件
  val newlyCreated = file.createNewFile()
  // 使文件可以被随机访问
  val raf = if (writable) new RandomAccessFile(file, "rw") else new RandomAccessFile(file, "r")
  try {
    // 将文件在磁盘上预先分配到最大大小
    if(newlyCreated) {
      if(maxIndexSize < entrySize)
        throw new IllegalArgumentException("Invalid max index size: " + maxIndexSize)
      raf.setLength(roundDownToExactMultiple(maxIndexSize, entrySize))
    }

    // 创建MappedByteBuffer，并将其的将读写指针定位到最后一个索引项
    _length = raf.length()
    val idx = {
      if (writable)
        raf.getChannel.map(FileChannel.MapMode.READ_WRITE, 0, _length)
      else
        raf.getChannel.map(FileChannel.MapMode.READ_ONLY, 0, _length)
    }
    if(newlyCreated)
      idx.position(0)
    else
      idx.position(roundDownToExactMultiple(idx.limit(), entrySize))
    // 返回该MappedByteBuffer
    idx
  } finally {
    CoreUtils.swallow(raf.close(), AbstractIndex)
  }
}

// 查找指定值的 lower bound 和 upper bound，即指定值的左右相邻的两个值
// lower bound 即小于等于指定值的最大值
// upper bound 即大于等于指定值的最小值
private def indexSlotRangeFor(
  // 待查找的索引文件
  idx: ByteBuffer,
  // 指定值
  target: Long,
  // 该值为索引项的key还是value
  searchEntity: IndexSearchEntity
): (Int, Int) = {
  if(_entries == 0)
    return (-1, -1)

  // 创建二分查找函数，方便后续调用
  def binarySearch(begin: Int, end: Int) : (Int, Int) = {
    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)
  }

  // 如果指定值处于索引文件的热区域，则搜索热区域
  val firstHotEntry = Math.max(0, _entries - 1 - _warmEntries)
  if(compareIndexEntry(parseEntry(idx, firstHotEntry), target, searchEntity) < 0) {
    return binarySearch(firstHotEntry, _entries - 1)
  }

  if(compareIndexEntry(parseEntry(idx, 0), target, searchEntity) > 0)
    return (-1, 0)

  // 否则搜索冷区域
  binarySearch(0, firstHotEntry)
}

// 获取第n个索引项
// 由于具体的索引类型不同，交给子类实现
protected def parseEntry(buffer: ByteBuffer, n: Int): IndexEntry

def resize(newSize: Int): Boolean = {
  inLock(lock) {
    val roundedNewSize = roundDownToExactMultiple(newSize, entrySize)

    if (_length == roundedNewSize) {
      debug(s"Index ${file.getAbsolutePath} was not resized because it already has size $roundedNewSize")
      false
    } else {
      // 使文件可以随机访问
      val raf = new RandomAccessFile(file, "rw")
      try {
        val position = mmap.position()

        // Windows不允许文件在被mmap的时候改变大小，需要先取消映射
        if (OperatingSystem.IS_WINDOWS)
          safeForceUnmap()
        // 改变文件大小
        raf.setLength(roundedNewSize)
        _length = roundedNewSize
        // 重新mmap到新的大小
        mmap = raf.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, roundedNewSize)
        _maxEntries = mmap.limit() / entrySize
        mmap.position(position)
        debug(s"Resized ${file.getAbsolutePath} to $roundedNewSize, position is ${mmap.position()} " +
          s"and limit is ${mmap.limit()}")
        true
      } finally {
        CoreUtils.swallow(raf.close(), AbstractIndex)
      }
    }
  }
}

// 8 = 4 (相对偏移量大小) + 4 (文件物理位置)
override def entrySize = 8

override protected def parseEntry(buffer: ByteBuffer, n: Int): OffsetPosition = {
  OffsetPosition(baseOffset + relativeOffset(buffer, n), physical(buffer, n))
}

// 读取第n个索引项的key（偏移量）
private def relativeOffset(buffer: ByteBuffer, n: Int): Int = buffer.getInt(n * entrySize)

// 读取第n个索引项的value（物理位置）
private def physical(buffer: ByteBuffer, n: Int): Int = buffer.getInt(n * entrySize + 4)

def append(offset: Long, position: Int): Unit = {
  inLock(lock) {
    require(!isFull, "Attempt to append to a full index (size = " + _entries + ").")
    if (_entries == 0 || offset > _lastOffset) {
      // 向文件尾增加一个索引项
      mmap.putInt(relativeOffset(offset))
      mmap.putInt(position)
      _entries += 1
      _lastOffset = offset
      // 校验当前读写位置必须是索引项的整数倍
      require(_entries * entrySize == mmap.position(), entries + " entries but file position in index is " + mmap.position() + ".")
    } else {
      // 索引项的key必须是单调递增的，否则直接抛异常
      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) {
    // 这里的duplicate相当于复制了一份mmap的position、limit和mark，而mmap的内容是共享的
    // 这是为了不影响mmap的写入，当lookup结束之后，原来的mmap的这几个内部变量都没发生改变
    val idx = mmap.duplicate
    val slot = largestLowerBoundSlotFor(idx, targetOffset, IndexSearchType.KEY)
    if(slot == -1)
      OffsetPosition(baseOffset, 0) // 注意这里当找不到lower bound时的返回值
    else
      parseEntry(idx, slot)
  }
}

def fetchUpperBoundOffset(fetchOffset: OffsetPosition, fetchSize: Int): Option[OffsetPosition] = {
  maybeLock(lock) {
    val idx = mmap.duplicate
    val slot = smallestUpperBoundSlotFor(idx, fetchOffset.position + fetchSize, IndexSearchType.VALUE)
    if (slot == -1)
      None // 注意这里当找不到upper bound时的返回值
    else
      Some(parseEntry(idx, slot))
  }
}