Apache Kafka源码分析 – Log Management

   1: /**

   2:  * An on-disk message set. An optional start and end position can be applied to the message set

   3:  * which will allow slicing a subset of the file.

   4:  * @param file The file name for the underlying log data

   5:  * @param channel the underlying file channel used

   6:  * @param start A lower bound on the absolute position in the file from which the message set begins

   7:  * @param end The upper bound on the absolute position in the file at which the message set ends

   8:  * @param isSlice Should the start and end parameters be used for slicing?

   9:  */

  10: @nonthreadsafe

  11: class FileMessageSet private[kafka](@volatile var file: File,

  12:                                     private[log] val channel: FileChannel,

  13:                                     private[log] val start: Int,

  14:                                     private[log] val end: Int,

  15:                                     isSlice: Boolean) extends MessageSet with Logging {...}

   1: /**

   2:  * An index that maps offsets to physical file locations for a particular log segment. This index may be sparse:

   3:  * that is it may not hold an entry for all messages in the log.

   4:  * 

   5:  * The index is stored in a file that is pre-allocated to hold a fixed maximum number of 8-byte entries.

   6:  * 

   7:  * The index supports lookups against a memory-map of this file. These lookups are done using a simple binary search variant

   8:  * to locate the offset/location pair for the greatest offset less than or equal to the target offset.

   9:  * 

  10:  * Index files can be opened in two ways: either as an empty, mutable index that allows appends or

  11:  * an immutable read-only index file that has previously been populated. The makeReadOnly method will turn a mutable file into an 

  12:  * immutable one and truncate off any extra bytes. This is done when the index file is rolled over.

  13:  * 

  14:  * No attempt is made to checksum the contents of this file, in the event of a crash it is rebuilt.

  15:  * 

  16:  * The file format is a series of entries. The physical format is a 4 byte "relative" offset and a 4 byte file location for the 

  17:  * message with that offset. The offset stored is relative to the base offset of the index file. So, for example,

  18:  * if the base offset was 50, then the offset 55 would be stored as 5. Using relative offsets in this way let's us use

  19:  * only 4 bytes for the offset.

  20:  * 

  21:  * The frequency of entries is up to the user of this class.

  22:  * 

  23:  * All external APIs translate from relative offsets to full offsets, so users of this class do not interact with the internal 

  24:  * storage format.

  25:  */

  26: class OffsetIndex(@volatile var file: File, val baseOffset: Long, val maxIndexSize: Int = -1) extends Logging {

  27:   private val lock = new ReentrantLock  //操作index文件需要加锁

  28:   

  29:   /* initialize the memory mapping for this index */

  30:   private var mmap: MappedByteBuffer =  //使用MappedByteBuffer来操作index文件以应对大文件

  31:     {

  32:       val newlyCreated = file.createNewFile()

  33:       val raf = new RandomAccessFile(file, "rw")

  34:       val len = raf.length()

  35:       val idx = raf.getChannel.map(FileChannel.MapMode.READ_WRITE, 0, len)          

  36:     }

  37:  

  38:   //通过byte偏移从buffer中读出某个entry的内容，offset和physical地址

  39:   /* return the nth offset relative to the base offset */

  40:   private def relativeOffset(buffer: ByteBuffer, n: Int): Int = buffer.getInt(n * 8)

  41:   /* return the nth physical position */

  42:   private def physical(buffer: ByteBuffer, n: Int): Int = buffer.getInt(n * 8 + 4)

  43:  

  44:   //通过二分查找找到targetOffset或最接近的offset（less than）

  45:   /**

  46:    * Find the largest offset less than or equal to the given targetOffset 

  47:    * and return a pair holding this offset and it's corresponding physical file position.

  48:    * 

  49:    * @param targetOffset The offset to look up.

  50:    * 

  51:    * @return The offset found and the corresponding file position for this offset. 

  52:    * If the target offset is smaller than the least entry in the index (or the index is empty),

  53:    * the pair (baseOffset, 0) is returned.

  54:    */

  55:   def lookup(targetOffset: Long): OffsetPosition = {...}

  56:  

  57: /**

  58:  * Get the nth offset mapping from the index

  59:  * @param n The entry number in the index

  60:  * @return The offset/position pair at that entry

  61:  */

  62: def entry(n: Int): OffsetPosition = {

  63:   maybeLock(lock) {

  64:     if(n >= entries)

  65:       throw new IllegalArgumentException("Attempt to fetch the %dth entry from an index of size %d.".format(n, entries))

  66:     val idx = mmap.duplicate

  67:     OffsetPosition(relativeOffset(idx, n), physical(idx, n))

  68:   }

  69: }

  70:  

  71: /**

  72:  * Append an entry for the given offset/location pair to the index. This entry must have a larger offset than all subsequent entries.

  73:  */

  74: def append(offset: Long, position: Int) {

  75:   inLock(lock) {

  76:     require(!isFull, "Attempt to append to a full index (size = " + size + ").")

  77:     if (size.get == 0 || offset > lastOffset) {

  78:       debug("Adding index entry %d => %d to %s.".format(offset, position, file.getName))

  79:       this.mmap.putInt((offset - baseOffset).toInt)

  80:       this.mmap.putInt(position)

  81:       this.size.incrementAndGet()

  82:       this.lastOffset = offset

  83:       require(entries * 8 == mmap.position, entries + " entries but file position in index is " + mmap.position + ".")

  84:     } else {

  85:       throw new InvalidOffsetException("Attempt to append an offset (%d) to position %d no larger than the last offset appended (%d) to %s."

  86:         .format(offset, entries, lastOffset, file.getAbsolutePath))

  87:     }

  88:   }

  89: }