上一篇【第45篇】Kafka日志存储源码解析四——FileMessageSet与ByteBufferMessageSet下一篇【第47篇】Kafka延迟操作DelayedOperation源码解析——优雅处理等待响应摘要如果说Log是日志段的管理者那么LogManager就是整个Kafka日志存储的总管家。它负责加载所有分区日志、调度日志清理/压缩任务、协调日志刷写、处理日志恢复。本文深入解析LogManager的架构设计、核心方法、以及它如何像交响乐指挥一样协调Log、LogSegment、OffsetIndex等组件高效工作。一、LogManager在Kafka存储架构中的位置1.1 从快递分拣中心理解LogManager┌─────────────────────────────────────────────────────────────┐ │ Kafka 存储架构全景图 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────┐ │ │ │ LogManager │ ← 本篇主角 │ │ │ (大管家) │ │ │ └──────┬──────┘ │ │ │ │ │ ┌─────────────┼─────────────┐ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Log实例1 │ │ Log实例2 │ │ Log实例3 │ ← 每个分区一个Log│ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │LogSegment│ │LogSegment│ │LogSegment│ ← 每个Log有多个│ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │OffsetIndex│ │OffsetIndex│ │OffsetIndex│ ← 每个Segment有│ │ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ │ 核心职责加载/恢复/清理/压缩/刷写 │ │ │ └─────────────────────────────────────────────────────────────┘LogManager就像快递分拣中心的总经理早上开门加载所有分区日志Log.load()日常运营调度清理线程CleanerThread、压缩线程CompressorThread确保不丢件定期刷写日志到磁盘flush()晚上盘点关闭所有日志释放资源shutdown()二、LogManager的架构设计2.1 核心字段一览// kafka/log/LogManager.scalaclassLogManager(logDirs:Seq[File],// 日志存储目录可配置多个提高并行度topicConfigs:Map[String,LogConfig],// 每个Topic的配置defaultConfig:LogConfig,// 默认配置cleanerConfig:CleanerConfig,// 日志压缩器配置ioThreads:Int,// IO线程数用于日志刷写heartbeat:Long,// 心跳间隔用于检测日志目录健康状态valrecoveryPoint:Long0// 恢复点崩溃恢复时用)extendsLoggingwithBrokerRole{// 核心数据结构1所有分区日志ConcurrentHashMap支持并发访问privatevallogsnewPool[TopicPartition,Log]()// 核心数据结构2日志清理调度器定时清理过期日志privatevallogCleaner:LogCleanerif(cleanerConfig.enableCleaner){newLogCleaner(cleanerConfig,logs)}else{null}// 核心数据结构3日志压缩调度器定时压缩日志段privatevallogCompressor:LogCompressornewLogCompressor(logs,defaultConfig)// 核心数据结构4定时任务调度器ScheduledExecutorServiceprivatevalscheduler:ScheduledExecutorServiceExecutors.newScheduledThreadPool(ioThreads)}2.2 启动流程LogManager.startup()┌─────────────────────────────────────────────────────────────┐ │ LogManager.startup() 启动流程 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 1: 加载所有日志目录log.dirs │ │ │ │ │ │ │ for (logDir - logDirs) { │ │ │ │ val dir new File(logDir) │ │ │ │ if (!dir.exists()) dir.mkdirs() │ │ │ │ │ │ │ │ // 加载该目录下所有分区日志 │ │ │ │ loadLogsInDir(dir) │ │ │ │ } │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 2: 恢复未刷写的日志崩溃恢复 │ │ │ │ │ │ │ for (log - logs.values()) { │ │ │ │ log.recover(recoveryPoint) │ │ │ │ } │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 3: 启动日志清理器LogCleaner │ │ │ │ │ │ │ if (logCleaner ! null) { │ │ │ │ logCleaner.start() │ │ │ │ } │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 4: 启动定时任务日志刷写/清理 │ │ │ │ │ │ │ scheduler.scheduleAtFixedRate( │ │ │ │ () flushDirtyLogs(), │ │ │ │ heartbeat, heartbeat, MILLISECONDS) │ │ │ │ │ │ │ │ scheduler.scheduleAtFixedRate( │ │ │ │ () cleanupExpiredLogs(), │ │ │ │ heartbeat, heartbeat, MILLISECONDS) │ │ │ └────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘三、核心功能一日志加载与恢复3.1 日志加载loadLogsInDir()// kafka/log/LogManager.scaladefloadLogsInDir(logDir:File):Unit{// 1. 列出目录下所有分区目录命名格式topic-uuidvaltopicPartitionslogDir.listFiles().filter(_.isDirectory())for(dir-topicPartitions){try{// 2. 解析目录名获取Topic和Partition编号valtopicPartitionparseTopicPartition(dir.getName())// 3. 创建Log实例核心vallogLog(dirdir,configtopicConfigs.getOrElse(topicPartition.topic,defaultConfig),schedulerscheduler,timetime)// 4. 加入内存中的logs池logs.put(topicPartition,log)}catch{casee:Exceptionerror(sFailed to load log$dir,e)}}}3.2 崩溃恢复Log.recover()┌─────────────────────────────────────────────────────────────┐ │ 日志恢复流程崩溃后重启 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ 场景Kafka宕机Page Cache中有未刷写的数据 │ │ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 1: 读取恢复点recoveryPoint │ │ │ │ │ │ │ 从上次成功刷写的位置开始恢复 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 2: 遍历所有日志段LogSegment │ │ │ │ │ │ │ for (segment - log.segments) { │ │ │ │ // 验证每个消息的CRC32校验和 │ │ │ │ segment.recover() │ │ │ │ } │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 3: 截断非法数据CRC校验失败 │ │ │ │ │ │ │ 将最后一个合法offset之后的数据全部删除 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 4: 更新恢复点recoveryPoint │ │ │ │ │ │ │ 记录本次成功恢复的位置 │ │ │ └────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘四、核心功能二日志清理策略Kafka支持两种日志清理策略删除Delete和压缩Compact。4.1 删除策略DefaultLogCleaner┌─────────────────────────────────────────────────────────────┐ │ 日志删除策略基于时间/大小 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ 触发条件满足任一即删除 │ │ │ │ 1. 时间策略log.retention.hours1687天 │ │ → 删除修改时间超过7天的日志段 │ │ │ │ 2. 大小策略log.retention.bytes10737418241GB │ │ → 从最老的日志段开始删除直到总大小1GB │ │ │ │ 删除流程 │ │ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 1: 计算需要删除的日志段 │ │ │ │ │ │ │ val deletableSegments log.segments.filter { │ │ │ │ segment │ │ │ │ // 时间策略 │ │ │ │ (now - segment.lastModified) retentionMs │ │ │ │ // 大小策略 │ │ │ │ || (totalSize - segment.size) retentionBytes │ │ │ │ } │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 2: 异步删除避免阻塞IO线程 │ │ │ │ │ │ │ scheduler.schedule( │ │ │ │ () deletableSegments.foreach(_.delete()) │ │ │ │ ) │ │ │ └────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘4.2 压缩策略LogCompactor┌─────────────────────────────────────────────────────────────┐ │ 日志压缩策略保留最新K/V │ ├─────────────────────────────────────────────────────────────┤ │ │ │ 核心思想对于每个Key只保留最新的一条消息 │ │ │ │ 压缩前 压缩后 │ │ ┌──────┬──────┬──────┬──────┐ ┌──────┐ │ │ │ Key1 │ Key2 │ Key1 │ Key3 │ │ Key2 │ │ │ │ Val1 │ Val2 │ Val3 │ Val4 │ → │ Key1 │ │ │ └──────┴──────┴──────┴──────┘ │ Key3 │ │ │ ↑ 同一Key的最新值 └──────┘ │ │ │ │ 压缩流程 │ │ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 1: 构建Key映射表Map[Key, Offset]│ │ │ │ │ │ │ 遍历所有日志段记录每个Key的最新offset│ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 2: 重写日志段只保留最新K/V │ │ │ │ │ │ │ 创建新的日志段写入去重后的消息 │ │ │ └────────────────┬───────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌────────────────────────────────────────────┐ │ │ │ Step 3: 替换旧日志段原子操作 │ │ │ │ │ │ │ 用新日志段替换旧的释放磁盘空间 │ │ │ └────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────┘压缩策略的适用场景✅Kafka Connect保留最新的Connector配置✅事件溯源Event Sourcing只关心实体最新状态❌日志收集不需要保留Key历史五、核心功能三日志刷写机制5.1 为什么要刷写Flush┌─────────────────────────────────────────────────────────────┐ │ 刷写平衡性能与可靠性 │ ├─────────────────────────────────────────────────────────────┤ │ │ │ 不刷写依赖Page Cache │ │ ┌──────┐ ┌──────┐ │ │ │ 磁盘 │ ◄── Page Cache ◄──│ Kafka │ │ │ └──────┘ └──────┘ │ │ ↑ 宕机时Page Cache数据丢失 │ │ │ │ 同步刷写每次写入都fsync │ │ ┌──────┐ ┌──────┐ │ │ │ 磁盘 │ ◄── fsync ◄──│ Kafka │ │ │ └──────┘ └──────┘ │ │ ↑ 性能极差磁盘IO成为瓶颈 │ │ │ │ Kafka的方案定时刷写平衡性能与可靠性 │ │ ┌──────┐ ┌──────┐ │ │ │ 磁盘 │ ◄── fsync ◄──│ Kafka │ │ │ └──────┘ 定时 └──────┘ │ │ ↑ 定期刷写既保证性能又避免大量数据丢失 │ │ │ └─────────────────────────────────────────────────────────────┘5.2 LogManager.flushDirtyLogs()// kafka/log/LogManager.scaladefflushDirtyLogs():Unit{// 遍历所有分区日志for((topicPartition,log)-logs){try{// 获取该分区最后一个刷写点valtimeSinceLastFlushtime.milliseconds()-log.lastFlushTime// 判断是否需要刷写log.flush.interval.msif(timeSinceLastFlushlog.config.flushInterval){// 关键只刷写脏数据Page Cache中有但磁盘没有的数据valflushedMessageslog.flush()info(sFlushed$flushedMessagesmessages for$topicPartition)}}catch{casee:Exceptionerror(sFailed to flush log$topicPartition,e)}}}六、性能调优LogManager相关参数参数默认值推荐值作用log.retention.hours168 (7天)24-72日志保留时间根据磁盘容量调整log.retention.bytes-1 (无限)1073741824 (1GB)日志保留大小防止磁盘写满log.segment.bytes1073741824 (1GB)268435456 (256MB)日志段大小调小可加快压缩/清理速度log.flush.interval.msLong.MaxValue (不刷写)60000 (1分钟)刷写间隔根据可靠性要求调整num.io.threads816-32IO线程数磁盘IO密集型场景调大log.cleaner.enablefalsetrue是否启用日志压缩事件溯源场景开启log.cleaner.threads12-4压缩线程数压缩密集型场景调大七、本篇小结今日收获LogManager是大管家负责加载/恢复/清理/压缩/刷写所有分区日志两种清理策略删除基于时间/大小和压缩保留最新K/V崩溃恢复机制通过recoveryPoint CRC校验确保数据不丢失定时刷写平衡性能与可靠性避免Page Cache数据丢失下篇预告文章047将深入DelayedOperation——Kafka的延迟操作 scheduler。当你发送消息时设置了acksall服务端必须等待所有副本拉取到数据后才能返回成功这个等待就是用DelayedOperation实现的上一篇【第45篇】Kafka日志存储源码解析四——FileMessageSet与ByteBufferMessageSet下一篇【第47篇】Kafka延迟操作DelayedOperation源码解析——优雅处理等待响应