kafka之二日志管理

本文主要内容有:

  1. 日志管理相关实现类
  2. 日志刷盘
  3. 日志清理
  4. checkpoint

日志管理相关实现类


kafka的日志管理是通过LogManager类实现的,它的主要作用是日志的创建、检索、清理。

kafkaServer启动LogManager线程

LogManager线程是在kafka节点服务启动的时候启动的,代码如下:

//kafka.server.KafkaServer
def startup() {
 try {
 info("starting")
 /* start log manager */
 //创建LogManager对象
 logManager = createLogManager(zkUtils.zkClient, brokerState)
 //启动LogManager线程
 logManager.startup()
 }
 catch {
 case e: Throwable =>
 fatal("Fatal error during KafkaServer startup. Prepare to shutdown", e)
 isStartingUp.set(false)
 shutdown()
 throw e
 }
}

说明,初始化LogManager的过程中会加载所有日志分区,对应方法为loadLogs(),在加载日志分区过程中会调用loadSegments()加载该分区所有的Segment文件,使用是的线程池进行加载的。

LogManager启动

在所有分区日志都加载完成后,KafkaServer调用startup()方法启动LogManager线程,在这个过程中会启动四个定时任务。

def startup() {
 /* Schedule the cleanup task to delete old logs */
 if(scheduler != null) {
 //note: 定时清理过期的日志 segment,并维护日志的大小
 info("Starting log cleanup with a period of %d ms.".format(retentionCheckMs))
 scheduler.schedule("kafka-log-retention",
 cleanupLogs,
 delay = InitialTaskDelayMs,
 period = retentionCheckMs,
 TimeUnit.MILLISECONDS)
 //note: 定时刷新还没有写到磁盘上日志
 info("Starting log flusher with a default period of %d ms.".format(flushCheckMs))
 scheduler.schedule("kafka-log-flusher",
 flushDirtyLogs,
 delay = InitialTaskDelayMs,
 period = flushCheckMs,
 TimeUnit.MILLISECONDS)
 //note: 定时将所有数据目录所有日志的检查点写到检查点文件中
 scheduler.schedule("kafka-recovery-point-checkpoint",
 checkpointRecoveryPointOffsets,
 delay = InitialTaskDelayMs,
 period = flushCheckpointMs,
 TimeUnit.MILLISECONDS)
 //note: 定时删除标记为 delete 的日志文件
 scheduler.schedule("kafka-delete-logs",
 deleteLogs,
 delay = InitialTaskDelayMs,
 period = defaultConfig.fileDeleteDelayMs,
 TimeUnit.MILLISECONDS)
 }
 //note: 如果设置为 true, 自动清理 compaction 类型的 topic
 if(cleanerConfig.enableCleaner)
 cleaner.startup()
}

四个定时任务:

  1. cleanupLogs,定时清理过期日志segment,并维护日志大小,默认5min执行一次
  2. flushDirtyLogs,定时刷新还没写到磁盘上数据
  3. checkpointRecoveryPointOffsets,定时将所有日志的checkpoint写到checkpoint文件中,默认60s执行一次
  4. deleteLogs,定时删除标记为delete的日志文件,默认30s执行一次。

checkpoint文件
在LogManager中有一个非常重要的文件---checkpoint文件:

  1. 创建LogManager时会读取checkpoint文件,并将每个分区对应的checkpoint作为日志的恢复点(recoveryPoint),最后创建分区对应的日志实例
  2. 在将日志刷盘时,将最新的偏移量作为日志的checkpoint进行更新
  3. LogManager启动一个定时任务,定时读取所有日志的检查点,并写入全局的检查点文件

日志刷盘


在linux系统中,当数据写入到文件系统后,数据其实在操作系统的page cache里,只有执行了刷盘后数据才会写到磁盘里。
在上面提到的定时任务flushDirtyLogs里,会定时将页面缓存中的数据刷新到磁盘中,kafka的刷盘策略有两种:

  1. 时间策略,通过log.flush.interval.ms进行配置,默认为无限大。
  2. 大小策略,通过log.flush.interval.messages进行配置,当数据超过这个值时进行刷盘。

需要提一下的是,定时任务里只会根据时间策略进行判断是否刷盘,根据大小判断是在append追加日志时进行的判断:

def append(records: MemoryRecords, assignOffsets: Boolean = true): LogAppendInfo = {
 // now append to the log
 segment.append(firstOffset = appendInfo.firstOffset,
 largestOffset = appendInfo.lastOffset,
 largestTimestamp = appendInfo.maxTimestamp,
 shallowOffsetOfMaxTimestamp = appendInfo.offsetOfMaxTimestamp,
 records = validRecords)
 // increment the log end offset
 updateLogEndOffset(appendInfo.lastOffset + 1)
 trace("Appended message set to log %s with first offset: %d, next offset: %d, and messages: %s"
 .format(this.name, appendInfo.firstOffset, nextOffsetMetadata.messageOffset, validRecords))
 if (unflushedMessages >= config.flushInterval)
 flush()
}

日志清理


为了保证分区总大小不超过阈值(log.retention.bytes),LogManager会定时清理旧数据。不过一般情况下是通过配置log.retention.hours来配置segment的保存时间的。

清理旧日志主要有两种:

  1. 删除,超过时间或大小阈值的旧segment直接进行删除
  2. 压缩,不是删除在,是是采用合并压缩的方式进行

checkpoint

kafka会对三类位移做checkpoint(高版本中还有一个Cleaner Offset)

  1. Log Start Offset
  2. Recovery Point Offset
  3. Replication Offset

Log Start Offset
每个topic partition log对象都有一个重要的位移字段:log start offset,用来标识分区消息对外部用户或应用可见的最早消息位移。在一些事件发生时kafka会触发对该值的更新。kafka对该offset进行checkpoint是为了更快保存分区的元数据,这样下次再初始化Log对象的时候能够直接加载并初始化log start offset.

Recovery Point Offset
它保存的是第一第未flush到磁盘的消息,kafka对它进行checkpoint能够加速日志希恢复的速度,因为直接从recovery point offset所在的日志段开始恢复即可,没必要从头恢复。

Replication Offset
它保存replication过程中副本的高水位(HW)位移值,通常场景是当副本重启后创建Log对象时可以直接使用这个文件中的offset对高水位对象进行赋值,省去了读取日志段然后再计算HW值的步骤.

checkpoint的主要作用是将kafka broker端重要的日志数据保存下来,下次可以快速的找到对应的数据。

参考文章:kafka中针对消费者位移的checkpoint机制具体是如何运转的?

作者:步履不停原文地址:https://segmentfault.com/a/1190000042236843

%s 个评论

要回复文章请先登录注册