存储机制

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 配置

# 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 策略

# 基于时间删除（默认 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 实现

// 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. 顺序写：追加写入，高吞吐