深入理解Kafka服务端之索引文件及mmap内存映射

//代码走这里说明newOffset正常
val offsetIdxFile = offsetIndexFile(dir, newOffset)//生成index文件
val timeIdxFile = timeIndexFile(dir, newOffset)//生成timeindex文件
val txnIdxFile = transactionIndexFile(dir, newOffset)//生成txnindex文件
//如果文件已存在则删除
for (file <- List(logFile, offsetIdxFile, timeIdxFile, txnIdxFile) if file.exists) {
  warn(s"Newly rolled segment file ${file.getAbsolutePath} already exists; deleting it first")
  Files.delete(file.toPath)
}


Option(segments.lastEntry).foreach(_.getValue.onBecomeInactiveSegment())

Option(segments.lastEntry).foreach(_.getValue.onBecomeInactiveSegment())

def onBecomeInactiveSegment() {
  timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, skipFullCheck = true)
  offsetIndex.trimToValidSize()
  timeIndex.trimToValidSize()
  log.trim()
}

public void trim() throws IOException {
    truncateTo(sizeInBytes());
}

def trimToValidSize() {
  inLock(lock) {
    //这里的_entries就是这个索引文件中索引项的数量
    //entrySize是索引项占用字节数。偏移量索引占用8字节，时间戳索引占用12字节
    resize(entrySize * _entries)
  }
}

/**
   * Reset the size of the memory map and the underneath file. This is used in two kinds of cases: (1) in
   * trimToValidSize() which is called at closing the segment or new segment being rolled; (2) at
   * loading segments from disk or truncating back to an old segment where a new log segment became active;
   * we want to reset the index size to maximum index size to avoid rolling new segment.
   */
def resize(newSize: Int): Boolean = {
  inLock(lock) {
    //将给定的newSize调整为entrySize的整数倍
    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 {
        //获取内存映射MappedByteBuffer的position
        val position = mmap.position()


        /* Windows won't let us modify the file length while the file is mmapped :-( */
        if (OperatingSystem.IS_WINDOWS)
          safeForceUnmap()
        //设置文件大小
        raf.setLength(roundedNewSize)
        _length = roundedNewSize
        //根据实际大小调整mmap内存映射的大小
        mmap = raf.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, roundedNewSize)
        //计算entry数量
        _maxEntries = mmap.limit()  entrySize
        //更新position值
        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)
      }
    }
  }
}

作用：重置内存映射和对应文件的大小。
这个方法在两种情况下使用：
(1)在trimToValidSize()方法中，该方法在关闭正在滚动的日志段或生成新日志段时被调用
(2)在从磁盘加载日志段或者一个新的日志段成为active截断旧日志段时调用
此时会重新调整索引文件的大小，避免因为索引文件大小达到阈值而滚动生成新日志段

val segment = LogSegment.open(dir,
          baseOffset = newOffset,
          config,
          time = time,
          fileAlreadyExists = false,
          initFileSize = initFileSize,
          preallocate = config.preallocate)

def open(dir: File, baseOffset: Long, config: LogConfig, time: Time, fileAlreadyExists: Boolean = false,
         initFileSize: Int = 0, preallocate: Boolean = false, fileSuffix: String = ""): LogSegment = {
  //这里设置了索引文件的大小，由参数：segment.index.bytes 决定，默认为10M
  val maxIndexSize = config.maxIndexSize
  new LogSegment(
    FileRecords.open(Log.logFile(dir, baseOffset, fileSuffix), fileAlreadyExists, initFileSize, preallocate),
    LazyIndex.forOffset(Log.offsetIndexFile(dir, baseOffset, fileSuffix), baseOffset = baseOffset, maxIndexSize = maxIndexSize),
    LazyIndex.forTime(Log.timeIndexFile(dir, baseOffset, fileSuffix), baseOffset = baseOffset, maxIndexSize = maxIndexSize),
    new TransactionIndex(baseOffset, Log.transactionIndexFile(dir, baseOffset, fileSuffix)),
    baseOffset,
    indexIntervalBytes = config.indexInterval,
    rollJitterMs = config.randomSegmentJitter,
    time)
}

LazyIndex.forOffset(Log.offsetIndexFile(dir, baseOffset, fileSuffix), baseOffset = baseOffset, maxIndexSize = maxIndexSize),
LazyIndex.forTime(Log.timeIndexFile(dir, baseOffset, fileSuffix), baseOffset = baseOffset, maxIndexSize = maxIndexSize),

//偏移量索引文件索引项
override def entrySize = 8
//时间戳索引文件索引项
override def entrySize = 12

def relativeOffset(offset: Long): Int = {
  val relativeOffset = toRelative(offset)
  if (relativeOffset.isEmpty)
    throw new IndexOffsetOverflowException(s"Integer overflow for offset: $offset (${file.getAbsoluteFile})")
  relativeOffset.get
}

private def toRelative(offset: Long): Option[Int] = {
  val relativeOffset = offset - baseOffset
  if (relativeOffset < 0 || relativeOffset > Int.MaxValue)
    None
  else
    Some(relativeOffset.toInt)
}

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

override def parseEntry(buffer: ByteBuffer, n: Int): TimestampOffset = {
  TimestampOffset(timestamp(buffer, n), baseOffset + relativeOffset(buffer, n))
}

  def lookup(targetOffset: Long): OffsetPosition = {
    maybeLock(lock) {
      //复制出整个索引映射区
      val idx = mmap.duplicate
      // largestLowerBoundSlotFor方法底层使用了改进版的二分查找算法寻找对应的槽
      val slot = largestLowerBoundSlotFor(idx, targetOffset, IndexSearchType.KEY)
      // 如果没找到，返回一个空的位置，即物理文件位置从0开始，表示从头读日志文件  
      // 否则返回slot槽对应的索引项
      if(slot == -1)
        OffsetPosition(baseOffset, 0)
      else
        parseEntry(idx, slot)
    }
  }

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)
  }
}

def findOffsetByTimestamp(timestamp: Long, startingOffset: Long = baseOffset): Option[TimestampAndOffset] = {
  val timestampOffset = timeIndex.lookup(timestamp)
  val position = offsetIndex.lookup(math.max(timestampOffset.offset, startingOffset)).position


  Option(log.searchForTimestamp(timestamp, position, startingOffset))
}

def append(offset: Long, position: Int): Unit = {
  inLock(lock) {
    // 索引文件如果已经写满，直接抛出异常
    require(!isFull, "Attempt to append to a full index (size = " + _entries + ").")
    // 要保证待写入的位移值offset比当前索引文件中所有现存的位移值都要大 
    // 这主要是为了维护索引的单调增加性
    if (_entries == 0 || offset > _lastOffset) {
      trace(s"Adding index entry $offset => $position to ${file.getAbsolutePath}")
      // 向mmap写入相对位移值
      mmap.putInt(relativeOffset(offset))
      // 向mmap写入物理文件位置
      mmap.putInt(position)
      //更新索引项数量
      _entries += 1
      // 更新当前索引文件最大位移值
      _lastOffset = offset
      // 确保写入索引项格式符合要求
      require(_entries * entrySize == mmap.position(), s"$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 maybeAppend(timestamp: Long, offset: Long, skipFullCheck: Boolean = false) {
  inLock(lock) {
    //如果索引文件已经写满，抛出异常
    if (!skipFullCheck)
      require(!isFull, "Attempt to append to a full time index (size = " + _entries + ").")
    // 确保相对偏移量单调增加性
    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}.")


    if (timestamp > lastEntry.timestamp) {
      trace(s"Adding index entry $timestamp => $offset to ${file.getAbsolutePath}.")
      // 向mmap写入时间戳
      mmap.putLong(timestamp)
      // 向mmap写入相对偏移量
      mmap.putInt(relativeOffset(offset))
      //更新entry的数量
      _entries += 1
      // 更新当前最新的索引项
      _lastEntry = TimestampOffset(timestamp, offset)
      require(_entries * entrySize == mmap.position(), _entries + " entries but file position in index is " + mmap.position() + ".")
    }
  }
}

privatedef truncateToEntries(entries: Int): Unit = {
  inLock(lock) {
    _entries = entries
    mmap.position(_entries * entrySize)
    _lastOffset = lastEntry.offset
    debug(s"Truncated index ${file.getAbsolutePath} to $entries entries;" +
      s" position is now ${mmap.position()} and last offset is now ${_lastOffset}")
  }
}

public class MappedByteBufferTest {
    public static void main(String[] args) {
        File file = new File("/opt/module/mmap.txt");
        long len = file.length();


        try {
            MappedByteBuffer mappedByteBuffer = new RandomAccessFile(file, "r")
                    //获取FileChannel
                    .getChannel()
                    //进行映射，文件类型为只读
                    .map(FileChannel.MapMode.READ_ONLY, 0, len);
                   ...
        } catch (IOException e) {}
    }
}