Kafka日志压缩

class LogManager(val logDirs: Array[File]){
  val logs = new Pool[TopicAndPartition, Log]()
  val recoveryPointCheckpoints=logDirs.map((_,new OffsetCheckpoint(new File(_,"checkpoint"))))

  loadLogs() //启动LogManager实例时,如果已经存在日志文件，要把它们加载到内存中

  private def loadLogs(): Unit = {
    val threadPools = mutable.ArrayBuffer.empty[ExecutorService]
    for (dir <- this.logDirs) { //按照log.dirs创建线程池，如果只配置一个目录就只有一个线程池
      val pool = Executors.newFixedThreadPool(ioThreads)
      threadPools.append(pool)
      //checkpoint文件一个日志目录只有一个，并不是每个Partition级别!
      //既然所有Partition公共一个checkpoint文件，那么文件内容当然要有Partition信息
      var recoveryPoints:Map[TopicAndPartition,Long]=recoveryPointCheckpoints(dir).read
      val jobsForDir = for {
        dirContent <- Option(dir.listFiles).toList //日志目录下的所有文件/文件夹
        logDir <- dirContent if logDir.isDirectory //Partition文件夹,忽略日志目录下的文件
      } yield {
        CoreUtils.runnable { //每个Partition文件夹创建一个线程，由线程池执行
          val topicPartition = Log.parseTopicPartitionName(logDir)
          val config = topicConfigs.getOrElse(topicPartition.topic, defaultConfig)
          val logRecoveryPoint = recoveryPoints.getOrElse(topicPartition, 0L) //分区的恢复点
          val current = new Log(logDir, config, logRecoveryPoint, scheduler, time) //恢复Log
          this.logs.put(topicPartition, current) //这里放入logs集合中，所有分区的Log满血复活
        }
      }
      jobsForDir.map(pool.submit).toSeq //提交任务
    }
  }

  //只有调用loadLogs后，logs才有值，后面的操作都依赖于logs
  def allLogs(): Iterable[Log] = logs.values
  def logsByDir = logs.groupBy{case (_,log)=>log.dir.getParent}
  val cleaner: LogCleaner = new LogCleaner(cleanerConfig,logDirs,logs)

  def startup() {
    scheduler.schedule("log-retention", cleanupLogs)
    scheduler.schedule("log-flusher", flushDirtyLogs)
    scheduler.schedule("recovery-point-checkpoint",checkpointRecoveryPointOffsets)
    if(cleanerConfig.enableCleaner) cleaner.startup()
  }
}

//日志文件刷写任务
private def flushDirtyLogs() = {
  for ((topicAndPartition, log) <- logs) {
    val timeSinceLastFlush = time.milliseconds - log.lastFlushTime
    if(timeSinceLastFlush >= log.config.flushMs) log.flush
  }
}
//checkpoint文件刷写任务
def checkpointRecoveryPointOffsets() {
  this.logDirs.foreach(checkpointLogsInDir)
}
private def checkpointLogsInDir(dir: File): Unit = {
  val recoveryPoints = this.logsByDir.get(dir.toString) //checkpoint是log.dirs目录级别
  //logsByDir对于每个dir都有多个Partition对应的Log，所以mapValues对每个Log获取recoveryPoint
  recoveryPointCheckpoints(dir).write(recoveryPoints.get.mapValues(_.recoveryPoint))
}

//只有flush的时候才会更新恢复点，不过flush并不是每次写都会发生的
def flush(offset: Long) : Unit = {
  if (offset <= this.recoveryPoint) return
  for(segment<-logSegments(this.recoveryPoint,offset)) //选择恢复点和当前之间的Segment
    segment.flush() //会分别刷写log数据文件和index索引文件（调用底层的fsync）
  if(offset > this.recoveryPoint) {
    this.recoveryPoint = offset //recoveryPoint实际上是offset
    lastflushedTime.set(time.milliseconds)
  }
}

//日志清理任务
def cleanupLogs() {
  for(log <- allLogs; if !log.config.compact)
    cleanupExpiredSegments(log) + cleanupSegmentsToMaintainSize(log)
}
private def cleanupExpiredSegments(log: Log): Int = {
  log.deleteOldSegments(time.milliseconds-_.lastModified>log.config.retentionMs)
}
private def cleanupSegmentsToMaintainSize(log: Log): Int = {
  var diff = log.size - log.config.retentionSize
  def shouldDelete(segment: LogSegment) = {
    if(diff - segment.size >= 0) {
      diff -= segment.size
      true
    } else false
  }
  log.deleteOldSegments(shouldDelete)
}

def deleteOldSegments(predicate: LogSegment => Boolean): Int = {
  //logSegments是Log的所有Segment，s是每个Segment
  val deletable = logSegments.takeWhile(s => predicate(s))
  if(segments.size == numToDelete) roll()
  deletable.foreach(deleteSegment(_))
}
private def deleteSegment(segment: LogSegment) {
  segments.remove(segment.baseOffset) //删除数据结构
  asyncDeleteSegment(segment) //异步删除Segment
}
private def asyncDeleteSegment(segment: LogSegment) {
  segment.changeFileSuffixes("", Log.DeletedFileSuffix)
  def deleteSeg() = segment.delete() //和flush一样最后调用log和index.delete
  scheduler.schedule("delete-file", deleteSeg)
}

case class LogToClean(topicPartition: TopicAndPartition, log: Log, 
    firstDirtyOffset: Long) extends Ordered[LogToClean] {
  val cleanBytes = log.logSegments(-1, firstDirtyOffset).map(_.size).sum
  val dirtyBytes = log.logSegments(firstDirtyOffset, 
    math.max(firstDirtyOffset, log.activeSegment.baseOffset)).map(_.size).sum
  val cleanableRatio = dirtyBytes / totalBytes.toDouble
  def totalBytes = cleanBytes + dirtyBytes
  override def compare(th:LogToClean)=math.signum(this.cleanableRatio-th.cleanableRatio)
}