前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Kafka的日志管理模块--LogManagerKafka源码分析-汇总

Kafka的日志管理模块--LogManagerKafka源码分析-汇总

作者头像
扫帚的影子
发布2018-09-05 17:07:21
8150
发布2018-09-05 17:07:21
举报
  • 这里说的日志不是为了追踪程序运行而打的日志,指的是Kafka接受到消息后将消息写入磁盘或从磁盘读取的子系统;
  • 它负责Log的创建,遍历,清理,读写等;
  • LogManager统领所有的Log对象, 具体的读写操作还是要转给Log对象,Log对象又包含若干个LogSegment, 一层套一层,逐层分解;
  • 它支持将本地的多个文件夹作出日志的存储目录;

LogManager
  • 所在文件:core/src/main/scala/kafka/log/LogManager.scala
  • LogManager的创建:
    1. 在KafkaServer启动时创建,通过调用 `KafkaServer.createLogManager实现。
  1. 每个Topic都可以单独设置自己Log的过期时间,roll大小等,这些信息存储在zk上,因此集群管理员可以通过调整zk上的相应配置,在不重启整个集群的前提下,动态调整这些信息;
  • LogManager的初始化:
  1. private val logs = new Pool[TopicAndPartition, Log](): 使用Pool管理所有的Log对象;
  2. createAndValidateLogDirs(logDirs): 目前支持将本地的多个文件夹作出日志的存储目录,因为需要创建和验证这些目录的有效性, 我们来看下是如何作的:
代码语言:javascript
复制
if(dirs.map(_.getCanonicalPath).toSet.size < dirs.size)
      throw new KafkaException("Duplicate log directory found: " + logDirs.mkString(", "))
    for(dir <- dirs) {
      if(!dir.exists) {
        info("Log directory '" + dir.getAbsolutePath + "' not found, creating it.")
        val created = dir.mkdirs()
        if(!created)
          throw new KafkaException("Failed to create data directory " + dir.getAbsolutePath)
      }
      if(!dir.isDirectory || !dir.canRead)
        throw new KafkaException(dir.getAbsolutePath + " is not a readable log directory.")
    }

判断是否有重复的log目录; 目录如不存在,则创建; 目录是否可读;

  1. val dirLocks = lockLogDirs(logDirs):使用文件锁锁定目录
代码语言:javascript
复制
dirs.map { dir =>
      val lock = new FileLock(new File(dir, LockFile))
      if(!lock.tryLock())
        throw new KafkaException("Failed to acquire lock on file .lock in " + lock.file.getParentFile.getAbsolutePath + 
                               ". A Kafka instance in another process or thread is using this directory.")
      lock
    }
  1. recoveryPointCheckpoints = logDirs.map(dir => (dir, new OffsetCheckpoint(new File(dir, RecoveryPointCheckpointFile)))).toMap: 创建每个目录中的recovery-point-offset-checkpoint文件(这个文件里记录的各个offset之前的数据均已落盘成功)的读取类对象;
  2. def loadLogs(): Unit: 恢复并且加载日志目录中的日志文件, 针对每个LogDir分别处理
代码语言:javascript
复制
val threadPools = mutable.ArrayBuffer.empty[ExecutorService]
    val jobs = mutable.Map.empty[File, Seq[Future[_]]]

    for (dir <- this.logDirs) {
      val pool = Executors.newFixedThreadPool(ioThreads)
      threadPools.append(pool)

      val cleanShutdownFile = new File(dir, Log.CleanShutdownFile)

      if (cleanShutdownFile.exists) {
        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)
      }

      var recoveryPoints = Map[TopicAndPartition, Long]()
      try {
        recoveryPoints = this.recoveryPointCheckpoints(dir).read
      } catch {
        case e: Exception => {
          warn("Error occured while reading recovery-point-offset-checkpoint file of directory " + dir, e)
          warn("Resetting the recovery checkpoint to 0")
        }
      }

      val jobsForDir = for {
        dirContent <- Option(dir.listFiles).toList
        logDir <- dirContent if logDir.isDirectory
      } yield {
        CoreUtils.runnable {
          debug("Loading log '" + logDir.getName + "'")

          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)
          val previous = this.logs.put(topicPartition, current)

          if (previous != null) {
            throw new IllegalArgumentException(
              "Duplicate log directories found: %s, %s!".format(
              current.dir.getAbsolutePath, previous.dir.getAbsolutePath))
          }
        }
      }

      jobs(cleanShutdownFile) = jobsForDir.map(pool.submit).toSeq
    }
    try {
      for ((cleanShutdownFile, dirJobs) <- jobs) {
        dirJobs.foreach(_.get)
        cleanShutdownFile.delete()
      }
    } catch {
      case e: ExecutionException => {
      }
    } finally {
      threadPools.foreach(_.shutdown())
    }
    info("Logs loading complete.")
  }

a. 如果kafka进程是优雅干净地退出的,会创建一个名为.kafka_cleanshutdown的文件作为标识; b. 启动kafka时, 如果不存在该文件, 则broker的状态进入到 RecoveringFromUncleanShutdown c. 针对dir下的每个topic子目录, 创建Log对象, 此对象在创建过程中会加载,恢复实际的消息, 每个这样的过程跑在一个使用**CoreUtils.runnable **创建的Job里, job再提交到线程池执行, 实际上是生成一个Feture, d. 等待c中所有的job都执行完, 以便完成所有的log加载,恢复过程;

  1. def startup(): 启动一个LogManager, 实际上是启动若干个定时任务:
代码语言:javascript
复制
scheduler.schedule("kafka-log-retention", 
                         cleanupLogs, 
                         delay = InitialTaskDelayMs, 
                         period = retentionCheckMs, 
                         TimeUnit.MILLISECONDS)
      scheduler.schedule("kafka-log-flusher", 
                         flushDirtyLogs, 
                         delay = InitialTaskDelayMs, 
                         period = flushCheckMs, 
                         TimeUnit.MILLISECONDS)
      scheduler.schedule("kafka-recovery-point-checkpoint",
                         checkpointRecoveryPointOffsets,
                         delay = InitialTaskDelayMs,
                         period = flushCheckpointMs,
                         TimeUnit.MILLISECONDS)
    }
    if(cleanerConfig.enableCleaner)
      cleaner.startup()

我们来过一遍: a. checkpointRecoveryPointOffsets: 将每个Topic-Partition的recovery-point(这个值就是已经落盘的offset值,因为有些log可能还在pagecache里,没有落盘)写入到recovery-point文件; b. flushDirtyLogs: 针对每一个Log对象,如果flush时间到,就调用log->flush, 将pagecache中的消息落盘; c. cleanupLogs: 针对清除策略是删除而不是压缩的Log, 依照时间和文件大小作清理:

代码语言:javascript
复制
 for(log <- allLogs; if !log.config.compact) {
      debug("Garbage collecting '" + log.name + "'")
      total += cleanupExpiredSegments(log) + cleanupSegmentsToMaintainSize(log)
    }

e. cleaner.startup(): 对日志不是删除, 而是采取压缩策略, 后面会专门讲下这个;

  1. def createLog(topicAndPartition: TopicAndPartition, config: LogConfig): Log: 创建Log对象, 使用nextLogDir()来选取当前log所在的目录;
  2. def deleteLog(topicAndPartition: TopicAndPartition): 移除Log;
  3. def truncateTo(partitionAndOffsets: Map[TopicAndPartition, Long]): 截取log到指定的offset, 同时写recovery-point文件:
代码语言:javascript
复制
for ((topicAndPartition, truncateOffset) <- partitionAndOffsets) {
      val log = logs.get(topicAndPartition)
      // If the log does not exist, skip it
      if (log != null) {
        //May need to abort and pause the cleaning of the log, and resume after truncation is done.
        val needToStopCleaner: Boolean = (truncateOffset < log.activeSegment.baseOffset)
        if (needToStopCleaner && cleaner != null)
          cleaner.abortAndPauseCleaning(topicAndPartition)
        log.truncateTo(truncateOffset)
        if (needToStopCleaner && cleaner != null)
          cleaner.resumeCleaning(topicAndPartition)
      }
    }
    checkpointRecoveryPointOffsets()
大致LogManager的内容就这么多

Kafka源码分析-汇总

本文参与 腾讯云自媒体分享计划,分享自作者个人站点/博客。
原始发表:2017.04.18 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • Kafka源码分析-汇总
相关产品与服务
对象存储
对象存储(Cloud Object Storage,COS)是由腾讯云推出的无目录层次结构、无数据格式限制,可容纳海量数据且支持 HTTP/HTTPS 协议访问的分布式存储服务。腾讯云 COS 的存储桶空间无容量上限,无需分区管理,适用于 CDN 数据分发、数据万象处理或大数据计算与分析的数据湖等多种场景。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档