摘要上一篇 Log.append 六步流程中第四步更新稀疏索引只提了一句。这篇把它彻底拆开Kafka 每个 LogSegment 有 .index 和 .timeindex 两个稀疏索引都通过 MappedByteBuffer 内存映射读写。为什么写 .log 用 FileChannel、写 .index 却用 mmap每条索引只有 8 或 12 字节用相对 offset 省一半空间每写 4KB 数据才追加一条查找时直接在映射内存上做下界二分。本篇从 OffsetIndex 类结构、append 真实源码、lookup 二分实现、TimeIndex 特殊性到 mmap 生命周期与真实坑逐层拆解。关键词Kafka 稀疏索引、OffsetIndex、TimeIndex、MappedByteBuffer、内存映射、mmap、二分查找、LogSegment从一个反直觉的设计切入读 Kafka 源码时很容易注意到一个矛盾写 .log 数据文件用FileChannel.write写 .index 索引文件用MappedByteBuffermmap。同样是往磁盘写文件为什么两套机制答案在两种写入的特征差异.log 是大文件顺序写每次 append 写一批消息几 KB 到几 MBFileChannel 一次系统调用搞定.index 是小文件随机写每次只写 8 字节固定条目按_entries × 8精确位置追加如果用 FileChannel每次都要position()write(ByteBuffer.allocate(8))两次系统调用开销远大于写入本身。mmap 把文件映射到用户态内存后mmap.putInt()就是直接写内存地址没有系统调用、没有缓冲区分配。而且索引文件天生就小每段预分配 10MB实际通常几 MB完全放得进内存映射。01 OffsetIndex 类结构与 mmap 内存布局每个 LogSegment 目录下三个文件以 baseOffset 命名20 位补零.log 数据文件最大 1GBFileChannel 顺序写、.index 偏移索引offset → positionmmap、.timeindex 时间索引timestamp → offsetmmap。两个索引在段创建时通过FileChannel.map(READ_WRITE, 0, maxIndexSize)映射到用户态内存。OffsetIndex 核心字段classOffsetIndex(valfile:File,valbaseOffset:Long,valmaxIndexSize:Int){privatevalmmap:MappedByteBuffer...// 文件内存映射privateval_entriesnewAtomicInteger(0)// 已写入条目数privatevalmaxEntriesmaxIndexSize/8// 最大条目数}_entries用 AtomicInteger——虽然 append 全程持 segmentLock但 lookup 和 append 可能在不同线程需要可见性保证。maxEntries maxIndexSize / 8是硬上限超了抛 “Index is full”。mmap 内存中条目紧密排列无填充OffsetIndex 每条 8 字节4B 相对 offset 4B positionTimeIndex 每条 12 字节8B timestamp 4B 相对 offset。写入位置用_entries × 8精确计算无需文件指针。02 append 写入真实源码与调用链在LogSegment.append内部每写一批消息累加bytesSinceLastIndexEntry达到index.interval.bytes默认 4096才写一条索引if(bytesSinceLastIndexEntryindexIntervalBytes){valrelativeOffset(targetOffset-baseOffset).toInt offsetIndex.append(relativeOffset,physicalPosition)timeIndex.append(maxTimestamp,relativeOffset)bytesSinceLastIndexEntry0}两个细节relativeOffset是 Int 不是 Long——用相对 offset 后段内消息数不超 2^314 字节够用比存 8 字节绝对 offset 省一半空间。physicalPosition是消息在 .log 文件中的起始字节位置来自写入前的channel.position()查找时直接 seek。OffsetIndex.append 真实实现defappend(offset:Int,position:Int):Unit{require(!isFull,Attempt to append to a full index)if(_entries.get0)require(offsetlastOffset,Offsets not increasing in index)mmap.position(_entries.get*8)mmap.putInt(offset)mmap.putInt(position)_entries.incrementAndGet()}三行关键逻辑require(offset lastOffset)保证 offset 严格单调递增——二分查找的前提乱序直接抛异常mmap.position(_entries.get * 8)用条目数 × 8 精确定位连续两次putInt写内存无系统调用。TimeIndex.append 写 12 字节putLong(maxTimestamp) putInt(relativeOffset)。timestamp 取 RecordBatch 中的最大 timestamp因为批次内消息时间可能不同取最大值保证查找时定位到的条目一定在目标时间之前。03 lookup 查找mmap 内存上的下界二分查找核心是在 mmap 内存上做下界二分lowerBound——找最大的 ≤ 目标值的条目而非精确匹配。完整调用链segments 中floorEntry定位段 → 计算相对 offset → mmap 内存上二分 → 取出 position →channel.position(position)顺序扫描直到找到目标。二分核心实现deflookup(targetOffset:Int):OffsetPosition{varlo0;varhientries-1while(lohi){valmid(lohi1)1// 上中位数避免死循环valoffmmap.getInt(mid*8)// 直接读内存无系统调用if(offtargetOffset)lomidelsehimid-1}newOffsetPosition(baseOffsetmmap.getInt(lo*8),mmap.getInt(lo*84))}三个实现细节mid (lo hi 1) 1用上中位数——因为lo mid不是 mid1用下中位数在 lo 和 hi 相邻时会死循环mmap.getInt(mid * 8)直接从映射内存读 4 字节无系统调用、无缓冲区分配、无内核拷贝几万条索引的段二分只需 ~15 次内存读微秒级完成下界而非精确匹配——索引稀疏目标 offset 大概率不在索引中找 floor 后顺序扫描最多扫 4KB。04 TimeIndex 的三个特殊设计TimeIndex 共用 mmap 机制但 timestamp 处理有三个独特设计① timestamp 取批次最大值。append 时写的是 RecordBatch 中的最大 timestamp不是每条消息的 timestamp。批次内消息时间可能不同Producer 端设置取最大值保证查找时间点 T 时定位到的条目对应的 offset 一定 ≤ T 所在的消息。② timestamp 不要求严格单调。OffsetIndex.append 有require(offset lastOffset)校验但 TimeIndex 没有——因为 timestamp 由 Producer 端设置可能乱序时钟回拨、重试消息。查找时用下界二分即可不要求严格递增。③ 查找时需要二次定位。TimeIndex 只能定位到 offset还需 OffsetIndex 二次定位到 position再读消息验证 timestamp。因为 TimeIndex 稀疏且 timestamp 取批次最大值定位到的 offset 可能早于目标时间需顺序扫描到第一条 timestamp ≥ 目标的消息。05 mmap 生命周期预分配、写入、截断、刷盘四阶段闭环每步有具体源码操作① 预分配 mmap段创建时。roll 出新段时创建空索引文件RandomAccessFile.setLength(maxIndexSize)预分配 10MB再channel.map(READ_WRITE, 0, maxIndexSize)映射整个文件。mmap 一旦创建大小不可变必须在 map 前扩到最大否则mmap.putInt超文件大小抛BufferOverflowException。② 追加写入append 过程中。按 4KB 间隔追加条目直接操作 mmap 内存地址纯内存操作无系统调用写后立即可读。此阶段磁盘文件仍是 10MB预分配大小。③ 封闭截断段滚动时。channel.truncate(actualSize)截断到实际大小entries × 8释放预分配多余空间——这就是磁盘上 .index 通常只有几 MB 而非 10MB 的原因。截断后旧 mmap 仍映射原 10MB但封闭段只查不写旧 mmap 可正常读截断部分读出 0关闭时统一回收。④ 刷盘 force 关闭后台或关闭时。默认不主动 force靠 OS 回写。LogManager 后台 flush 任务或 Broker 关闭时mmap.force()落盘然后channel.close()mmap 由 GC 回收无显式 unmap API。06 mmap 的三个真实坑① force() 是全量 fsync开销大。mmap.force()对整个 10MB 映射区域执行 msync/fsync即使只写了几十字节。每次 append 都 force 性能会崩所以默认不主动 force靠 OS 回写只在后台和关闭时调用——和 .log 策略一致用副本机制兜底持久性。② unmap 依赖 GC段删除后内存不立即释放。MappedByteBuffer 无显式 unmap 方法JDK 内部有Unsafe.invokeCleaner但 Kafka 未用段删除后 mmap 可能还占内存直到 GC。高吞吐下段滚动频繁时可能短暂内存上涨这也是segment.bytes不宜设太小的原因之一。③ 截断后旧 mmap 映射仍有效。段封闭时channel.truncate()缩小文件但已存在的 MappedByteBuffer 仍映射原 10MB继续写超截断位置会抛 IOException。Kafka 做法封闭段后只查不写旧 mmap 正常读关闭时统一回收重新打开已封闭段索引时创建新对象重新 mmap 截断后的大小。写在最后Kafka 稀疏索引的 mmap 写入核心是一个朴素判断大文件顺序写用 FileChannel小文件随机写用 mmap各取所长。8 字节固定条目、相对 offset 省一半空间、4KB 间隔稀疏策略、预分配截断生命周期、mmap 内存上的下界二分——每个设计都有具体源码支撑和明确权衡。理解了索引的写入和查找Log.append 完整流程就通了写数据文件持久化消息写索引快速定位消息一主一辅共同支撑高吞吐写入和低延迟查找。
摘要上一篇 Log.append 六步流程中第四步更新稀疏索引只提了一句。这篇把它彻底拆开Kafka 每个 LogSegment 有 .index 和 .timeindex 两个稀疏索引都通过 MappedByteBuffer 内存映射读写。为什么写 .log 用 FileChannel、写 .index 却用 mmap每条索引只有 8 或 12 字节用相对 offset 省一半空间每写 4KB 数据才追加一条查找时直接在映射内存上做下界二分。本篇从 OffsetIndex 类结构、append 真实源码、lookup 二分实现、TimeIndex 特殊性到 mmap 生命周期与真实坑逐层拆解。关键词Kafka 稀疏索引、OffsetIndex、TimeIndex、MappedByteBuffer、内存映射、mmap、二分查找、LogSegment从一个反直觉的设计切入读 Kafka 源码时很容易注意到一个矛盾写 .log 数据文件用FileChannel.write写 .index 索引文件用MappedByteBuffermmap。同样是往磁盘写文件为什么两套机制答案在两种写入的特征差异.log 是大文件顺序写每次 append 写一批消息几 KB 到几 MBFileChannel 一次系统调用搞定.index 是小文件随机写每次只写 8 字节固定条目按_entries × 8精确位置追加如果用 FileChannel每次都要position()write(ByteBuffer.allocate(8))两次系统调用开销远大于写入本身。mmap 把文件映射到用户态内存后mmap.putInt()就是直接写内存地址没有系统调用、没有缓冲区分配。而且索引文件天生就小每段预分配 10MB实际通常几 MB完全放得进内存映射。01 OffsetIndex 类结构与 mmap 内存布局每个 LogSegment 目录下三个文件以 baseOffset 命名20 位补零.log 数据文件最大 1GBFileChannel 顺序写、.index 偏移索引offset → positionmmap、.timeindex 时间索引timestamp → offsetmmap。两个索引在段创建时通过FileChannel.map(READ_WRITE, 0, maxIndexSize)映射到用户态内存。OffsetIndex 核心字段classOffsetIndex(valfile:File,valbaseOffset:Long,valmaxIndexSize:Int){privatevalmmap:MappedByteBuffer...// 文件内存映射privateval_entriesnewAtomicInteger(0)// 已写入条目数privatevalmaxEntriesmaxIndexSize/8// 最大条目数}_entries用 AtomicInteger——虽然 append 全程持 segmentLock但 lookup 和 append 可能在不同线程需要可见性保证。maxEntries maxIndexSize / 8是硬上限超了抛 “Index is full”。mmap 内存中条目紧密排列无填充OffsetIndex 每条 8 字节4B 相对 offset 4B positionTimeIndex 每条 12 字节8B timestamp 4B 相对 offset。写入位置用_entries × 8精确计算无需文件指针。02 append 写入真实源码与调用链在LogSegment.append内部每写一批消息累加bytesSinceLastIndexEntry达到index.interval.bytes默认 4096才写一条索引if(bytesSinceLastIndexEntryindexIntervalBytes){valrelativeOffset(targetOffset-baseOffset).toInt offsetIndex.append(relativeOffset,physicalPosition)timeIndex.append(maxTimestamp,relativeOffset)bytesSinceLastIndexEntry0}两个细节relativeOffset是 Int 不是 Long——用相对 offset 后段内消息数不超 2^314 字节够用比存 8 字节绝对 offset 省一半空间。physicalPosition是消息在 .log 文件中的起始字节位置来自写入前的channel.position()查找时直接 seek。OffsetIndex.append 真实实现defappend(offset:Int,position:Int):Unit{require(!isFull,Attempt to append to a full index)if(_entries.get0)require(offsetlastOffset,Offsets not increasing in index)mmap.position(_entries.get*8)mmap.putInt(offset)mmap.putInt(position)_entries.incrementAndGet()}三行关键逻辑require(offset lastOffset)保证 offset 严格单调递增——二分查找的前提乱序直接抛异常mmap.position(_entries.get * 8)用条目数 × 8 精确定位连续两次putInt写内存无系统调用。TimeIndex.append 写 12 字节putLong(maxTimestamp) putInt(relativeOffset)。timestamp 取 RecordBatch 中的最大 timestamp因为批次内消息时间可能不同取最大值保证查找时定位到的条目一定在目标时间之前。03 lookup 查找mmap 内存上的下界二分查找核心是在 mmap 内存上做下界二分lowerBound——找最大的 ≤ 目标值的条目而非精确匹配。完整调用链segments 中floorEntry定位段 → 计算相对 offset → mmap 内存上二分 → 取出 position →channel.position(position)顺序扫描直到找到目标。二分核心实现deflookup(targetOffset:Int):OffsetPosition{varlo0;varhientries-1while(lohi){valmid(lohi1)1// 上中位数避免死循环valoffmmap.getInt(mid*8)// 直接读内存无系统调用if(offtargetOffset)lomidelsehimid-1}newOffsetPosition(baseOffsetmmap.getInt(lo*8),mmap.getInt(lo*84))}三个实现细节mid (lo hi 1) 1用上中位数——因为lo mid不是 mid1用下中位数在 lo 和 hi 相邻时会死循环mmap.getInt(mid * 8)直接从映射内存读 4 字节无系统调用、无缓冲区分配、无内核拷贝几万条索引的段二分只需 ~15 次内存读微秒级完成下界而非精确匹配——索引稀疏目标 offset 大概率不在索引中找 floor 后顺序扫描最多扫 4KB。04 TimeIndex 的三个特殊设计TimeIndex 共用 mmap 机制但 timestamp 处理有三个独特设计① timestamp 取批次最大值。append 时写的是 RecordBatch 中的最大 timestamp不是每条消息的 timestamp。批次内消息时间可能不同Producer 端设置取最大值保证查找时间点 T 时定位到的条目对应的 offset 一定 ≤ T 所在的消息。② timestamp 不要求严格单调。OffsetIndex.append 有require(offset lastOffset)校验但 TimeIndex 没有——因为 timestamp 由 Producer 端设置可能乱序时钟回拨、重试消息。查找时用下界二分即可不要求严格递增。③ 查找时需要二次定位。TimeIndex 只能定位到 offset还需 OffsetIndex 二次定位到 position再读消息验证 timestamp。因为 TimeIndex 稀疏且 timestamp 取批次最大值定位到的 offset 可能早于目标时间需顺序扫描到第一条 timestamp ≥ 目标的消息。05 mmap 生命周期预分配、写入、截断、刷盘四阶段闭环每步有具体源码操作① 预分配 mmap段创建时。roll 出新段时创建空索引文件RandomAccessFile.setLength(maxIndexSize)预分配 10MB再channel.map(READ_WRITE, 0, maxIndexSize)映射整个文件。mmap 一旦创建大小不可变必须在 map 前扩到最大否则mmap.putInt超文件大小抛BufferOverflowException。② 追加写入append 过程中。按 4KB 间隔追加条目直接操作 mmap 内存地址纯内存操作无系统调用写后立即可读。此阶段磁盘文件仍是 10MB预分配大小。③ 封闭截断段滚动时。channel.truncate(actualSize)截断到实际大小entries × 8释放预分配多余空间——这就是磁盘上 .index 通常只有几 MB 而非 10MB 的原因。截断后旧 mmap 仍映射原 10MB但封闭段只查不写旧 mmap 可正常读截断部分读出 0关闭时统一回收。④ 刷盘 force 关闭后台或关闭时。默认不主动 force靠 OS 回写。LogManager 后台 flush 任务或 Broker 关闭时mmap.force()落盘然后channel.close()mmap 由 GC 回收无显式 unmap API。06 mmap 的三个真实坑① force() 是全量 fsync开销大。mmap.force()对整个 10MB 映射区域执行 msync/fsync即使只写了几十字节。每次 append 都 force 性能会崩所以默认不主动 force靠 OS 回写只在后台和关闭时调用——和 .log 策略一致用副本机制兜底持久性。② unmap 依赖 GC段删除后内存不立即释放。MappedByteBuffer 无显式 unmap 方法JDK 内部有Unsafe.invokeCleaner但 Kafka 未用段删除后 mmap 可能还占内存直到 GC。高吞吐下段滚动频繁时可能短暂内存上涨这也是segment.bytes不宜设太小的原因之一。③ 截断后旧 mmap 映射仍有效。段封闭时channel.truncate()缩小文件但已存在的 MappedByteBuffer 仍映射原 10MB继续写超截断位置会抛 IOException。Kafka 做法封闭段后只查不写旧 mmap 正常读关闭时统一回收重新打开已封闭段索引时创建新对象重新 mmap 截断后的大小。写在最后Kafka 稀疏索引的 mmap 写入核心是一个朴素判断大文件顺序写用 FileChannel小文件随机写用 mmap各取所长。8 字节固定条目、相对 offset 省一半空间、4KB 间隔稀疏策略、预分配截断生命周期、mmap 内存上的下界二分——每个设计都有具体源码支撑和明确权衡。理解了索引的写入和查找Log.append 完整流程就通了写数据文件持久化消息写索引快速定位消息一主一辅共同支撑高吞吐写入和低延迟查找。
