# 存储机制 > Kafka 高性能的秘密之一在于其精妙的存储设计。理解 Log Segment、索引机制和零拷贝是深入掌握 Kafka 的关键。 ## 存储架构 ### 目录结构 ``` /kafka-logs/ ├── orders-0/ # Topic: orders, Partition: 0 │ ├── 00000000000000000000.log # 第一个 Segment │ ├── 00000000000000000000.index # 偏移量索引 │ ├── 00000000000000000000.timeindex # 时间戳索引 │ ├── 00000000000000123456.log # 第二个 Segment (baseOffset=123456) │ ├── 00000000000000123456.index │ ├── 00000000000000123456.timeindex │ └── leader-epoch-checkpoint ├── orders-1/ # Topic: orders, Partition: 1 │ └── ... └── __consumer_offsets-0/ # 内部 Topic: 消费位移 └── ... ``` ### 存储层次 ``` Topic └── Partition └── Log └── Segment ├── .log (数据文件) ├── .index (偏移量索引) └── .timeindex (时间索引) ``` ## Log Segment ### Segment 结构 ``` Segment 组成： ┌─────────────────────────────────────────────────────────┐ │ Log Segment │ │ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ .log 文件（数据） │ │ │ │ ┌────────┬────────┬────────┬────────┬───────┐ │ │ │ │ │Message │Message │Message │Message │ ... │ │ │ │ │ │ 0 │ 1 │ 2 │ 3 │ │ │ │ │ │ └────────┴────────┴────────┴────────┴───────┘ │ │ │ └─────────────────────────────────────────────────┘ │ │ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ .index 文件（偏移量索引） │ │ │ │ ┌──────────────┬──────────────┬─────────────┐ │ │ │ │ │ Offset → Pos │ Offset → Pos │ ... │ │ │ │ │ └──────────────┴──────────────┴─────────────┘ │ │ │ └─────────────────────────────────────────────────┘ │ │ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ .timeindex 文件（时间索引） │ │ │ │ ┌──────────────┬──────────────┬─────────────┐ │ │ │ │ │ Time → Offset│ Time → Offset│ ... │ │ │ │ │ └──────────────┴──────────────┴─────────────┘ │ │ │ └─────────────────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────────────────┘ ``` ### 消息格式 ``` Message (V2 格式，Kafka 0.11+) ┌──────────────────────────────────────────────────────┐ │ Record Batch │ ├──────────────────────────────────────────────────────┤ │ baseOffset (8 bytes) - 批次起始偏移量 │ │ batchLength (4 bytes) - 批次长度 │ │ partitionLeaderEpoch (4 bytes) │ │ magic (1 byte) - 消息格式版本 │ │ crc (4 bytes) - 校验和 │ │ attributes (2 bytes) - 压缩、时间戳类型等 │ │ lastOffsetDelta (4 bytes) - 最后一条消息的偏移增量 │ │ firstTimestamp (8 bytes) - 第一条消息时间戳 │ │ maxTimestamp (8 bytes) - 最大时间戳 │ │ producerId (8 bytes) - 生产者 ID │ │ producerEpoch (2 bytes) - 生产者 Epoch │ │ baseSequence (4 bytes) - 起始序列号 │ │ records count (4 bytes) - 消息数量 │ ├──────────────────────────────────────────────────────┤ │ Records │ │ ┌──────────────────────────────────────────────────┐ │ │ │ length (varint) - 记录长度 │ │ │ │ attributes (1 byte) - 属性 │ │ │ │ timestampDelta (varint) - 时间戳增量 │ │ │ │ offsetDelta (varint) - 偏移量增量 │ │ │ │ keyLength (varint) - Key 长度 │ │ │ │ key (bytes) - Key 数据 │ │ │ │ valueLength (varint) - Value 长度 │ │ │ │ value (bytes) - Value 数据 │ │ │ │ headers (array) - 消息头 │ │ │ └──────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────┘ ``` ### Segment 配置 ```properties # Segment 大小（默认 1GB） log.segment.bytes=1073741824 # Segment 滚动时间（默认 7 天） log.roll.hours=168 # 索引大小（默认 10MB） log.index.size.max.bytes=10485760 ``` ## 索引设计 ### 稀疏索引 Kafka 使用稀疏索引，不是每条消息都建立索引： ``` .log 文件 ┌────────────────────────────────────────────────────┐ │ Offset: 0 1 2 3 4 5 6 7 8 9 ... │ │ Pos: 0 50 100 150 200 250 300 350 400 450 ... │ └────────────────────────────────────────────────────┘ .index 文件（稀疏索引，每 4KB 数据建一个索引） ┌─────────────────────────────────────────────┐ │ Offset → Position │ │ 0 → 0 │ │ 3 → 150 （跳过了 1, 2） │ │ 6 → 300 （跳过了 4, 5） │ │ 9 → 450 （跳过了 7, 8） │ └─────────────────────────────────────────────┘ ``` ### 查找过程 ``` 查找 Offset=5 的消息： 1. 二分查找 Segment 文件 - 比较文件名（baseOffset） - 找到 00000000000000000000.log 2. 在 .index 中二分查找 - 找到 Offset=3 → Position=150 3. 在 .log 中顺序查找 - 从 Position=150 开始 - 顺序读取直到 Offset=5 ┌───────────────────────────────────────────────────────┐ │ 查找流程 │ │ │ │ Offset=5 │ │ │ │ │ ▼ │ │ ┌─────────────────┐ │ │ │ 二分查找 Segment │ │ │ │ (按文件名) │ │ │ └────────┬────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────┐ │ │ │ 二分查找 .index │ │ │ │ 找到 3→150 │ │ │ └────────┬────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────┐ │ │ │ 顺序扫描 .log │ │ │ │ 从 150 开始 │ │ │ │ 找到 Offset=5 │ │ │ └─────────────────┘ │ │ │ └───────────────────────────────────────────────────────┘ ``` ### 时间索引 ``` .timeindex 文件：时间戳 → Offset 用途： 1. 按时间查找消息 2. 日志清理（基于时间） 3. Consumer 按时间 seek 使用： consumer.offsetsForTimes(timestampsToSearch); ``` ## 日志清理 ### 两种清理策略 ``` 1. Delete（删除） - 删除过期的 Segment - 基于时间或大小 2. Compact（压缩） - 保留每个 Key 的最新值 - 适用于 changelog 类型的 Topic 配置： log.cleanup.policy=delete # 或 compact, delete,compact ``` ### Delete 策略 ```properties # 基于时间删除（默认 7 天） log.retention.hours=168 # 基于大小删除（默认无限制） log.retention.bytes=-1 # 检查间隔（默认 5 分钟） log.retention.check.interval.ms=300000 ``` ### Compact 策略 ``` Log Compaction 示意： 清理前： ┌────────────────────────────────────────────┐ │ K1:V1 K2:V1 K1:V2 K3:V1 K2:V2 K1:V3 K3:V2 │ └────────────────────────────────────────────┘ 清理后（保留每个 Key 的最新值）： ┌────────────────────┐ │ K1:V3 K2:V2 K3:V2 │ └────────────────────┘ 使用场景： - CDC 数据 - 状态存储 - KTable ``` ## 零拷贝 ### 传统数据传输 ``` 传统方式（4 次拷贝，4 次上下文切换）： ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 磁盘 │───▶│内核缓冲区│───▶│用户缓冲区│───▶│Socket缓冲│───▶ 网络 └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ │ DMA 拷贝 │ CPU 拷贝 │ CPU 拷贝 │ DMA 拷贝 │ │ │ │ └──────────────┴──────────────┴──────────────┘ read() write() 系统调用 系统调用 ``` ### 零拷贝（sendfile） ``` 零拷贝方式（2 次拷贝，2 次上下文切换）： ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 磁盘 │───▶│内核缓冲区│───▶│Socket缓冲│───▶ 网络 └──────────┘ └──────────┘ └──────────┘ │ │ │ │ DMA 拷贝 │ DMA 拷贝 │ │ │ │ └──────────────┴──────────────┘ sendfile() 系统调用 优势： 1. 减少 CPU 拷贝 2. 减少上下文切换 3. 减少内存占用 ``` ### Java 实现 ```java // Kafka 使用 FileChannel.transferTo() // 底层调用 sendfile 系统调用 FileChannel fileChannel = new FileInputStream(file).getChannel(); SocketChannel socketChannel = SocketChannel.open(); // 零拷贝传输 fileChannel.transferTo(position, count, socketChannel); ``` ## Page Cache ### Page Cache 的作用 ``` Kafka 大量使用操作系统的 Page Cache： ┌─────────────────────────────────────────────────────┐ │ 内存 │ │ ┌───────────────────────────────────────────────┐ │ │ │ Page Cache │ │ │ │ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │ │ │Page │ │Page │ │Page │ │Page │ │Page │ │ │ │ │ │ 1 │ │ 2 │ │ 3 │ │ 4 │ │ 5 │ │ │ │ │ └─────┘ └─────┘ └─────┘ └─────┘ └─────┘ │ │ │ └───────────────────────────────────────────────┘ │ │ ▲ │ │ │ │ │ ┌─────────────────────┴─────────────────────────┐ │ │ │ Kafka Broker │ │ │ │ (不在 JVM 堆中管理，避免 GC) │ │ │ └───────────────────────────────────────────────┘ │ │ │ │ └────────────────────────┼────────────────────────────┘ │ ▼ ┌─────────┐ │ 磁盘 │ └─────────┘ 优势： 1. 读写都经过 Page Cache 2. 热点数据自动缓存 3. 避免 JVM GC 影响 4. 进程重启后缓存仍有效 ``` ## 顺序写 ### 顺序写 vs 随机写 ``` 顺序写磁盘： ┌────────────────────────────────────────┐ │ │ │ → → → → → → → → → → → → → → → │ │ 追加写入，磁头不需要移动 │ │ │ └────────────────────────────────────────┘ 性能：~600MB/s 随机写磁盘： ┌────────────────────────────────────────┐ │ ↓ ↓ ↓ ↓ ↓ │ │ ┌──┴───┴───────┴────┴──────┴────┐ │ │ │ 随机位置写入，磁头频繁移动 │ │ │ └───────────────────────────────┘ │ └────────────────────────────────────────┘ 性能：~100KB/s Kafka 只追加写入，充分利用顺序 I/O 性能 ``` ## 面试高频问题 ### 1. Kafka 为什么性能这么高？ ``` 1. 顺序写磁盘 - 只追加写入，避免随机 I/O - 顺序 I/O 性能接近内存 2. 零拷贝 - sendfile 系统调用 - 减少 CPU 拷贝和上下文切换 3. Page Cache - 利用操作系统缓存 - 避免 JVM GC 4. 批量处理 - 批量发送、批量压缩 - 减少网络开销 5. 分区并行 - 多 Partition 并行处理 - 水平扩展 ``` ### 2. Kafka 的索引机制是怎样的？ ``` 稀疏索引： 1. 不是每条消息都建索引 2. 每隔一定字节建一个索引 3. 查找时先二分查找索引，再顺序扫描 两种索引： 1. .index：Offset → Position 2. .timeindex：Timestamp → Offset 查找过程： 1. 二分查找 Segment（按文件名） 2. 二分查找 Index（定位大致位置） 3. 顺序扫描 Log（精确定位） ``` ### 3. Log Compaction 的原理？ ``` 1. 保留每个 Key 的最新值 2. 删除旧版本的消息 3. 后台线程异步执行 4. 不影响正常读写 使用场景： - CDC 数据（数据库变更） - 状态存储（KTable） - 配置信息 配置： log.cleanup.policy=compact log.cleaner.enable=true ``` ### 4. 零拷贝是如何实现的？ ``` 传统方式： 磁盘 → 内核缓冲区 → 用户缓冲区 → Socket 缓冲区 → 网络 （4 次拷贝，4 次上下文切换） 零拷贝（sendfile）： 磁盘 → 内核缓冲区 → Socket 缓冲区 → 网络 （2 次拷贝，2 次上下文切换） Java 实现： FileChannel.transferTo() 底层调用 sendfile 系统调用 ``` ## 总结 ``` Kafka 存储机制要点： 1. Log Segment：分段存储，便于清理和查找 2. 稀疏索引：Offset 索引 + 时间索引 3. 日志清理：Delete 删除 + Compact 压缩 4. 零拷贝：sendfile 系统调用 5. Page Cache：利用 OS 缓存，避免 GC 6. 顺序写：追加写入，高吞吐 ```