kafka logManager类 kafka存储机制

  logManager类：管理kafka数据log的类，包括数据clean，flush等操作

   Log类：每个tplog的对象

      logSegment：每个tplog目录下的文件对象

          filemessageSet：每个log file的管道类

          base offset：在topic中的绝对offset值

          offsetindex：每个log index的管道map类，存储相对offset值和文件position

 

   按照partition分区topic，分发到各个机子上

   partition上有多个log文件，每个log文件一个索引文件

   log文件是实际的数据，索引文件是log文件里数据的相对偏移量和在log文件里的position，偏移量offset是一段数据生成一个offset，避免offset文件过大

 

1.初始化：

val RecoveryPointCheckpointFile = "recovery-point-offset-checkpoint"
  val LockFile = ".lock"
  val InitialTaskDelayMs = 30*1000
  private val logCreationOrDeletionLock = new Object
  private val logs = new Pool[TopicAndPartition, Log]()//所有log的对象,一个topicpartition 一个log对象

  //获得log文件，并获得文件channel锁
  createAndValidateLogDirs(logDirs)
  private val dirLocks = lockLogDirs(logDirs)
  private val recoveryPointCheckpoints = logDirs.map(dir => (dir, new OffsetCheckpoint(new File(dir, RecoveryPointCheckpointFile)))).toMap
  //遍历所有的log，生成Log对象，并且执行log clean（checkposition）
  loadLogs()

主要方法loadLogs：

if (cleanShutdownFile.exists) {//表示上次关闭kafka时，已经clean完，这次不需要clean
        debug(
          "Found clean shutdown file. " +
          "Skipping recovery for all logs in data directory: " +
          dir.getAbsolutePath)
      } else {
        // log recovery itself is being performed by `Log` class during initialization
        brokerState.newState(RecoveringFromUncleanShutdown)
      }

      //获得log下recover文件
      val recoveryPoints = this.recoveryPointCheckpoints(dir).read

      val jobsForDir = for {
        dirContent <- Option(dir.listFiles).toList
        logDir <- dirContent if logDir.isDirectory
      } yield {
        Utils.runnable {
          debug("Loading log '" + logDir.getName + "'")
          //从文件目录上获得topic和partition
          val topicPartition = Log.parseTopicPartitionName(logDir.getName)
          //从map中获得topic的自定义config，如果
          val config = topicConfigs.getOrElse(topicPartition.topic, defaultConfig)
          val logRecoveryPoint = recoveryPoints.getOrElse(topicPartition, 0L)

          val current = new Log(logDir, config, logRecoveryPoint, scheduler, time)
          val previous = this.logs.put(topicPartition, current)
          //判断是否有重复的topic+partition
          if (previous != null) {
            throw new IllegalArgumentException(
              "Duplicate log directories found: %s, %s!".format(
              current.dir.getAbsolutePath, previous.dir.getAbsolutePath))
          }
        }
      }
      //对每个logDir执行 上边的runnable，生成Log对象添加到log pool中
      jobs(cleanShutdownFile) = jobsForDir.map(pool.submit).toSeq

  其中new Log方法，为初始化log file和index

 主方法：loadSegments

   1.处理swap文件，log则重新加载（rename），index则删除

   2.加载log和index，恢复不存在的index

private def loadSegments() {
    // create the log directory if it doesn't exist
    dir.mkdirs()
    
    // first do a pass through the files in the log directory and remove any temporary files 
    // and complete any interrupted swap operations
    for(file <- dir.listFiles if file.isFile) {
      if(!file.canRead)
        throw new IOException("Could not read file " + file)
      val filename = file.getName
      if(filename.endsWith(DeletedFileSuffix) || filename.endsWith(CleanedFileSuffix)) {
        // if the file ends in .deleted or .cleaned, delete it
        file.delete()
      } else if(filename.endsWith(SwapFileSuffix)) {//文件用于swap时候，恢复log
        // we crashed in the middle of a swap operation, to recover:
        // if a log, swap it in and delete the .index file
        // if an index just delete it, it will be rebuilt
        //如果是index则删除，如果是log则重新加载（重命名），并删除已经存在的index
        val baseName = new File(Utils.replaceSuffix(file.getPath, SwapFileSuffix, ""))
        if(baseName.getPath.endsWith(IndexFileSuffix)) {
          file.delete()
        } else if(baseName.getPath.endsWith(LogFileSuffix)){
          // delete the index
          val index = new File(Utils.replaceSuffix(baseName.getPath, LogFileSuffix, IndexFileSuffix))
          index.delete()
          // complete the swap operation
          val renamed = file.renameTo(baseName)
          if(renamed)
            info("Found log file %s from interrupted swap operation, repairing.".format(file.getPath))
          else
            throw new KafkaException("Failed to rename file %s.".format(file.getPath))
        }
      }
    }

    // now do a second pass and load all the .log and .index files
    for(file <- dir.listFiles if file.isFile) {
      val filename = file.getName
      if(filename.endsWith(IndexFileSuffix)) {
        // if it is an index file, make sure it has a corresponding .log file 查看index log是否对应的 log，如果没有则删除
        val logFile = new File(file.getAbsolutePath.replace(IndexFileSuffix, LogFileSuffix))
        if(!logFile.exists) {
          warn("Found an orphaned index file, %s, with no corresponding log file.".format(file.getAbsolutePath))
          file.delete()
        }
      } else if(filename.endsWith(LogFileSuffix)) {
        // if its a log file, load the corresponding log segment
        // 文件名是start offset
        val start = filename.substring(0, filename.length - LogFileSuffix.length).toLong
        val hasIndex = Log.indexFilename(dir, start).exists
        //建立tplog中 每个日志文件对象 logsegment，包含filemessage，offsetindex，baseoffset值
        val segment = new LogSegment(dir = dir, 
                                     startOffset = start,
                                     indexIntervalBytes = config.indexInterval, 
                                     maxIndexSize = config.maxIndexSize,
                                     rollJitterMs = config.randomSegmentJitter,
                                     time = time)
        if(!hasIndex) {
          error("Could not find index file corresponding to log file %s, rebuilding index...".format(segment.log.file.getAbsolutePath))
          //重建index文件和内存索引,文件和内存索引是用的channel map机制
          segment.recover(config.maxMessageSize)
        }
        segments.put(start, segment)
      }
    }

    if(logSegments.size == 0) {
      // no existing segments, create a new mutable segment beginning at offset 0
      segments.put(0L, new LogSegment(dir = dir,
                                     startOffset = 0,
                                     indexIntervalBytes = config.indexInterval, 
                                     maxIndexSize = config.maxIndexSize,
                                     rollJitterMs = config.randomSegmentJitter,
                                     time = time))
    } else {
      recoverLog()
      // reset the index size of the currently active log segment to allow more entries
      activeSegment.index.resize(config.maxIndexSize)
    }

    // sanity check the index file of every segment to ensure we don't proceed with a corrupt segment
    for (s <- logSegments)
      s.index.sanityCheck()
  }

 -----------------------------初始化完毕---------------------------------

 

startup方法中三个功能：

1.cleanupLogs

2.flushDirtyLogs

3.checkpointRecoveryPointOffsets

 

1.cleanupLogs

 两个方法一个是超时(超时是modify时间)，一个是大小（大小是最老的小于diff）

  private def cleanupExpiredSegments(log: Log): Int = {
    val startMs = time.milliseconds
    //参数为log manager开始时间-tplog的修改时间 和 配置retention时间 比较，超过则需要删除，返回true
    //删除的是最后一次修改时间超过retention time的
    log.deleteOldSegments(startMs - _.lastModified > log.config.retentionMs)
  }

  /**
   * 删除规则，是tplog超过阈值，从最老的开始找，找到file的大小小于diff的时候删除
   * 如果当前log file大小大于diff，则停止（原则是等最后一个文件可删除）
   *  Runs through the log removing segments until the size of the log
   *  is at least logRetentionSize bytes in size
   */
  private def cleanupSegmentsToMaintainSize(log: Log): Int = {
    if(log.config.retentionSize < 0 || log.size < log.config.retentionSize)
      return 0//当配置小于0，或log大小小于配置
    var diff = log.size - log.config.retentionSize
    def shouldDelete(segment: LogSegment) = {
      if(diff - segment.size >= 0) {//如果需要删除的大小 大于或等于 logfile，则返回true
        diff -= segment.size
        true
      } else {
        false
      }
    }
    log.deleteOldSegments(shouldDelete)
  }

  参数：

  清理日志，距离上次修改时间大于config时间，则删除

  val logCleanupIntervalMs = props.getLongInRange("log.retention.check.interval.ms", 5*60*1000, (1, Long.MaxValue))

  log clean参数，达到log大小上限，log的position

  val logRetentionBytes = props.getLong("log.retention.bytes", -1)

 

  def deleteOldSegments(predicate: LogSegment => Boolean): Int = {
    // find any segments that match the user-supplied predicate UNLESS it is the final segment 
    // and it is empty (since we would just end up re-creating it
    val lastSegment = activeSegment
    //超时，并且包含segment，则删除，获得删除list segment
    val deletable = logSegments.takeWhile(s => predicate(s) && (s.baseOffset != lastSegment.baseOffset || s.size > 0))
    val numToDelete = deletable.size
    if(numToDelete > 0) {
      lock synchronized {
        // we must always have at least one segment, so if we are going to delete all the segments, create a new one first
        if(segments.size == numToDelete)
          roll()
        // remove the segments for lookups
        deletable.foreach(deleteSegment(_))//从segment集合中移除，修改文件名称为delete结尾，并异步删除
      }
    }
    numToDelete
  }

 

 2.flushDirtyLogs

flush的message条数和时间间隔
    /* the maximum time in ms that a message in any topic is kept in memory before flushed to disk */
  val logFlushIntervalMs = props.getLong("log.flush.interval.ms", logFlushSchedulerIntervalMs)
  
  /**
   * Flush any log which has exceeded its flush interval and has unwritten messages.
   */
  private def flushDirtyLogs() = {
    debug("Checking for dirty logs to flush...")

    for ((topicAndPartition, log) <- logs) {
      try {
        val timeSinceLastFlush = time.milliseconds - log.lastFlushTime
        debug("Checking if flush is needed on " + topicAndPartition.topic + " flush interval  " + log.config.flushMs +
              " last flushed " + log.lastFlushTime + " time since last flush: " + timeSinceLastFlush)
        if(timeSinceLastFlush >= log.config.flushMs)
          log.flush
      } catch {
        case e: Throwable =>
          error("Error flushing topic " + topicAndPartition.topic, e)
      }
    }
  }
  
    @threadsafe
  def flush() {
    LogFlushStats.logFlushTimer.time {
      log.flush()
      index.flush()
    }
  }

 

3.checkpointRecoveryPointOffsets

checkpointRecoveryPointOffsets，标记logdir上的恢复点，避免启动时，需要恢复所有log，生成index

 

是按照logdir遍历，logdir中包含多个tplog

 

  /**
   * Make a checkpoint for all logs in provided directory.
   */
  private def checkpointLogsInDir(dir: File): Unit = {
    //获得当前dir的所有tplog，value：Map【TopicAndPartition, Log】
    val recoveryPoints = this.logsByDir.get(dir.toString)
    if (recoveryPoints.isDefined) {
      //mapValues重新生成map的value，write参数（topicAndPartition：recoverPoint）；
      //write将tplog的offset写入recover文件的tmp文件中，删除旧文件，rename为recover文件 _是Log对象（value）
      this.recoveryPointCheckpoints(dir).write(recoveryPoints.get.mapValues(_.recoveryPoint))
    }
  }

 

 logmanager里实现log compact功能

    if(cleanerConfig.enableCleaner)
      cleaner.startup()//log compact