这个格式好点：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
遍历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对象读写消息
}
