Kafka代码走读-LogCleaner

 

这个格式好点：https://github.com/haogrgr/haogrgr-test/blob/master/logs/kafka_source.txt

5.日志管理器启动 logManager.startup(){
    主要是启动一些定时任务:
    
    a)LogManager.cleanupLogs, 根据配置清理日志文件(cleanupExpiredSegments[根据时间], cleanupSegmentsToMaintainSize[根据大小]).
    Log.deleteOldSegments {
        按时间和按大小逻辑相似, 按时间则是根据每一个Segment的最后修改时间判断; 按大小, 则是依次累加每一个LogSegment的大小, 当累计大小大于配置大小后的全部Segment都删除.
        注意, 这里不删除当前活动的Segment
         获取到要删除的Segment后, 依次调用kafka.log.Log.deleteSegment方法{
            首先将Segment从Log.segments中删除, 
            而后重命名Segment的index和log文件为.deleted后缀
            而后建立异步任务, 异步的删除Segment文件
        }
    }
    
    b)LogManager.flushDirtyLogs, 根据配置定时刷盘, 刷盘后, 会更新恢复点(log.recoveryPoint).
    LogManager.flushDirtyLogs{
        遍历logs
        val timeSinceLastFlush = time.milliseconds - log.lastFlushTime
        if(timeSinceLastFlush >= log.config.flushMs){
            log.flush
        }
    }
    Log.flush{
        经过treemap找到最新offset和上一次recoveryPoint(恢复点)的Segment, 依次调用Segment.flush方法, 内部调用index和logFile的force方法刷盘
        而后记录最新的recoveryPoint(恢复点), 更新lastflushedTime
    }
    
    c)LogManager.checkpointRecoveryPointOffsets, 根据配置, 将每一个Log.recoveryPoint刷盘, 建立checkpointRecoveryPointOffset文件, 用途前面2.4.1有说.
    
    而后, 启动cleaner线程
    cleaner.startup() {
        大概做用就是, 清理日志, 具体方法是, 对同一个key的消息, 保留offset最大的消息体, 其余的丢弃, 对于空消息体, 则表示删除.
        能够看到, 适合某些场景, 好比说用kafka来同步某个属性的值, 每变更一次, 就发一次消息, 消费者更新为最新值, 这种状况, 新的客户端对于某一key只须要获取最新的一条消息的值就好了.
        有点相似于redis的aof rewrite.
        而后有时候清理后, 日志文件变得很小, 就须要合并多个日志文件.
        
        由于清理过程须要大量的IO操做, 因此经过Throttler类来对限速, 防止对正常的写日志产生影响
        由于清理是一个长时间的过程, 且是多线程的, 因此须要有个地方标记这个分区正在清理, 因此就有了LogCleanerManager, 来存放状态信息
        
        内容比较多, 具体见5.1
    }
    
}

5.1.Cleaner线程 {
    
    根据配置, 建立LogCleaner, 若是cleanerConfig.enableCleaner == true
    new LogCleaner(cleanerConfig, logDirs, logs, time = time) {
        建立清理管理器, 主要负责维护分区在clean过程当中的状态信息
        private[log] val cleanerManager = new LogCleanerManager(logDirs, logs);
        
        限流器, 防止clean占用的IO可控, 线程安全, 多个清理线程共用一个.
        private val throttler = new Throttler(desiredRatePerSec = config.maxIoBytesPerSecond, checkIntervalMs = 300, throttleDown = true, "cleaner-io", "bytes", time = time)
        
        //清理线程
        val cleaners = (0 until config.numThreads).map(new CleanerThread(_))
    }
    
    先看和业务关系不大的Throttler{
        累加计数器, 而后判断时间是否是该检查速度了, 
        若是要检查速度, 就算出当前的速度, 和配置的数度对比, 看看是否是须要调整
        若是须要调整, 具体算法思路为下:
        具体问题也抽象为这样的问题, 两我的以不一样的速度走了一秒钟, 求, 一秒钟后, 快的人停多久, 才能等到慢的人.
        具体解法就是, 根据速度和时间, 算出距离差距, 而后经过   (慢人的速度 / 距离差距   = 慢的人要赶的时间  = 快的人要等的时间)
        
        这里kafka的实现相似, 不过省略了中间过程, 因此比较难看懂, 具体能够当作下面代码
        if(needAdjustment) {
            val desiredRateMs = desiredRatePerSec / Time.MsPerSec.toDouble //限制的速度(慢人速度)
            val elapsedMs = elapsedNs.toDouble / Time.NsPerMs              //已经跑了多久时间
            val diffBytes = observedSoFar - (desiredRateMs * elapsedMs)    //快人跑的距离   - (慢人速度  * 已经跑了多久时间  = 慢人跑的距离) = 差距
            val newSleepTime = round(diffBytes / desiredRateMs)            //差距   / 慢人速度  = 要等待的时间
            if(newSleepTime > 0) {
                time.sleep(newSleepTime)
            }
        }
        把中间过程化简后, 就是kafka的写法
    }
    
    在来看看维护清理状态的 new LogCleanerManager(logDirs, logs) {
        用来保存清理进度快照, 保存的是每一个分区上次清理到的位置(firstDirtyOffset), firstDirtyOffset以前的表示已经清理过了, 每次清理完成会更新快照文件.
        写:当clean完成的时候, 会调用LogCleanerManager.doneCleaning来更新分区最新的offset快照
        读:Clean线程中, 会不断经过LogCleanerManager.grabFilthiestLog()来获取脏日志, 而判断脏日志, 是经过获取到快照中记录的firstDirtyOffset, 从而计算出脏日志大小, 来判断是否须要清理.
        清理须要策略, 若是不记录清理进度, 则每次都作所有的扫描, 低效, 因此经过保存clean进度快照来记录清理进度, 减小没必要要的清理.
        val checkpoints = logDirs.map(dir => (dir, new OffsetCheckpoint(new File(dir, offsetCheckpointFile)))).toMap
        
        一个Map, 用来记录分区正在进行中的clean状态信息. 有三个状态(LogCleaningInProgress, LogCleaningAborted, LogCleaningPaused)
        当判断到分区脏日志大于阀值时, 会进入LogCleaningInProgress状态, 开始清理, 当清理完成时, 更新快照, 讲分区从inProgress中移除.
        LogCleaningInProgress => LogCleaningAborted : 日志截取(truncate)时, 须要等待Clean完成, 且开始日志截取后, 就不能进行Clean, 须要等待日志截取完成, 才能继续Clean, 这个是经过LogCleaningAborted状态来实现的
        过程是这样的, 当日志截取时, 若是正在进行Clean, 则LogCleaningInProgress => LogCleaningAborted, 而后等待状态到LogCleaningPaused, 当Clean完成时, 会进行LogCleaningAborted => LogCleaningPaused
        这样就实现了截取日志时, Clean已经完成, 且分区被标记为LogCleaningPaused状态, 这样Clean线程后续也会忽略这个分区, 完成日志截取后, 经过CleanerManager.resumeCleaning方法, 将分区状态信息从Map中移除, 等待下次clean
        一样, 删除Log时, 也要作相似操做, 这时, 先将Log实例从Log池中移除, 而后保证Clean已经完成, 由于Log已经再也不池中了, 因此下次Clean也会再Clean了, 因此这里用的是Cleaner.abortCleaning, 里面组合了abortAndPauseCleaning, resumeCleaning
        val inProgress = mutable.HashMap[TopicAndPartition, LogCleaningState]()
        
        好了, LogCleanerManager基本功能清楚了, 接下来回到LogCleaner
    }
    
    最好看看Clean线程 CleanerThread extends ShutdownableThread {
        ShutdownableThread里面的run方法里循环调用子类的doWork方法, 因此具体逻辑咱们看doWork方法
        
        应该是具体Clean操做逻辑的类
        val cleaner = new Cleaner() {
            清理过程当中的状态信息, 如开始时间, 读取字节数等信息, 一个用来表是当前的状态, 一个用来表示上一次完成的状态
            val statsUnderlying = (new CleanerStats(time), new CleanerStats(time))
            
            用来合并同Key消息的Map, 主要是索引用, 先读一遍脏日志, 创建Map
            offsetMap = new SkimpyOffsetMap(memory = math.min(config.dedupeBufferSize / config.numThreads, Int.MaxValue).toInt, hashAlgorithm = config.hashAlgorithm){
                既然是hash, 就先看hash算法吧
                private def positionOf(hash: Array[Byte], attempt: Int): Int = {
                    具体就是hash以后的值是一个byte数组, 当冲突次数小于(hashSize - 4)时, 取Int(hash_byte[attempt, attempt + 3]), 
                    当冲突次数高于(hashSize - 4)时, 取Int(hash_byte[12, hashSize]) + (attempt - hashSize + 4)
                    即先求hash, 而后冲突时, 使用简单的策略来继续新的hash值, 来计算新的位置
                }
                内部维护一个ByteBuffer, 不存储原始Key, 只存key的hash值, 碰撞后, 继续新的位置, 读取时相似, 读取, 比对, hash不相对, 计算新的hash, 再读取比对
                碰撞挺高的, 越到后面, 插入效率越低, 碰撞高了, 读取效率也较低, 不支持删除(1000元素的容量, 插满后, 碰撞率为19左右)
            }
            
        }
        
        doWork逻辑 cleanOrSleep {
            调用cleanerManager.grabFilthiestLog(), 扫描全部Log获取要清理的分区, 具体条件为: 1)Log配置开启compact, 2)没有处于inProgress中, 3)脏日志大于minCleanableRatio比例, 4)取最脏的那个.
            grabFilthiestLog返回LogToClean对象, 里面包含脏日志比例, 字节数, firstDirtyOffset等信息
            
            清理逻辑 cleaner.clean(cleanable) {
                清理Clean过程当中的统计信息stats.clear(), 准备开始新的Clean过程
                
                先构建出Key索引Map, 清理范围是[firstDirtyOffset, activeSegment.baseOffset], 即清理除活动segment外的段
                val upperBoundOffset = log.activeSegment.baseOffset
                val endOffset = buildOffsetMap(log, cleanable.firstDirtyOffset, upperBoundOffset, offsetMap) + 1 {
                    map.clear(), 清空offsetMap, 准备建索引
                    根据offsetMap容量和负载因子, 计算出最多能清理到的offset(minStopOffset)        
                    而后遍历segment, 读取文件(有限流), 构建map, buildOffsetMapForSegment(){
                        不细说, 具体就是读取文件内容到buffer, 而后经过ByteBufferMessageSet来遍历消息, 创建map, 若是buffer不够读一条消息的, 就增大, 继续读
                        返回索引的最后一条消息的offset, 处理完后, 会恢复buffer
                    }
                    返回索引的最后一条消息的offset, 加1以后, 就是真实的清理上界offset(不含自身)
                }
                
                val deleteHorizonMs = 计算一个时间点, 用来判断消息是否须要忽略(删除), 即压缩的过程同时作清理的工做, 根据配置, 获取已清理过的最大offset的文件的修改时间, 没有文件则取0
                
                而后将offset[0, endOffset]的segment拿出来处理(即0到offsetMap中已索引的最大的offset), 根据大小进行group操做, 由于之前可能进行过clean操做, 致使segment可能变小, 这里group成配置指定的大小, 再处理.
                group后的结构是一个List<List<Segment>>, 内层的List是要合并的segement
                
                遍历group, 对里面的每一个组作clean操做 cleanSegments(log, group, offsetMap, deleteHorizonMs){
                    接下来就是读老文件, 合并同key的offset, 在写到临时文件, 最后重命名文件, 删除老文件, 换成新文件, 具体4.1有提到
                    
                    先建立log和index的临时文件(.cleaned), 若是存在, 则表示上次处理到一半的文件, 直接删除, 处理过的消息, 先写到.clean文件中, 防止写到一半挂了, 保证一致完整性.
                    同时建立对应的OffsetIndex和LogSegment对象
                    
                    而后对每一个segment进行处理 cleanInto(log.topicAndPartition, old, cleaned, map, retainDeletes){
                        处理分区前, 先检查分区是否是处于LogCleaningAborted状态(truncate时会进入这个状态, 具体看5.1), 是就抛异常结束这里处理
                        
                        而后清理readbuffer, writebuffer, 而后借助ByteBufferMessageSet对象来读取消息, 中间会判断readbuffer是否是过小(1条msg都读不到), 过小就扩容和前面buildOffsetMap相似,
                        先吧消息读到readbuffer里面(即一次读多条, 不够一条消息就须要扩容), 再对于每条消息, 获取到key和offset, 而后根据前面构建的OffsetMap来判断这条消息是否是留(offset要大于或等于OffsetMap里面的(map里面是最大的offset)), 
                        也会判断这个文件是否是达到了删除的条件, 要删的文件, 则消息不用处理, 最后, 若是消息体为空, 表示要忽略这条消息
                        
                        将消息写到writebuffer里面去, 由于readbuffer和writebuffer同样大, 且扩容的时候一块儿扩容的, 因此不怕writebuffer大小不够, 而后将writebuffer写到segment中去
                        
                        最后还原readbuff, writebuffer, clean过程当中, 也会有相似于进度收集, IO限速等控制
                    }
                    
                    好了, 一组segment已经clean成一个了, 接着trim一下index文件到真实的大小, 而后flush日志和索引到文件, 接着恢复一下文件的lastModified, 由于不恢复会致使原本达到删除时间的文件继续存在
                    
                    最后, 交换一下, 用新的segment替换老的segment, log.replaceSegments(cleaned, segments){
                        具体4.1有提到, 先将新的segment文件的后缀从.cleaned重命名到.swap文件, 重命名成功后, 若是进程挂了, 启动时loadSegments会完成后续的操做
                        若是刚重命名为.swap, 老的log尚未删除掉, 则下次启动的时候, loadSegments中重命名会失败, 致使启动失败, 能够手动删除老的log文件再启动
                        
                        将新的segment放入到log.segments中, 将老的一个个移除(除了刚才put的, put新的seg至关于删除了这个老的seg, 因此这里不移除了), 
                        而后将老的文件重命名为.deleted文件, 异步线程中删除老的segment文件 , 最后, 将新的segment重命名, 去掉swap后缀, 完成clean
                    }
                    
                }cleanSegments
                
            } cleaner.clean
            
            日志记录  recordStats, 记录上次Clean状态this.lastStats = stats, 顺便交换Clean状态类(Cleaner.statsUnderlying.swap)
            好了, clean操做总数是结束了
            
        }cleanOrSleep
                
    }CleanerThread
    
}

6.LogManager总结{
    首先, kafka对于一个topic, 会分为多个partition, 一个partition一个文件夹, 分区下面又分为多个segment, segment中又分为log和index, .log文件是最终的消息存放文件
    
    log文件负责消息的读写, index负责index的读写, segment聚合index和log, 提供一个统一的读写接口, 屏蔽索引等相关的操做细节, 
    而Log聚合segment, 维护lastOffset, 统一flush操做, 屏蔽掉文件滚动操做的细节, 提供一个更加上层的接口, 屏蔽底层文件的读写.
    
    LogManager则负责提供快照, 日志Clean, 相关定时任务管理等功能, 外部经过LogManager来获取Log对象读写消息
}