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