1
# 磁盘性能基准测试
2
# 顺序写入
3
dd if=/dev/zero of=/var/lib/kafka/data/test bs=1M count=1024 oflag=direct
4

5
# 随机写入（使用 fio）
6
fio --name=randwrite --ioengine=libaio --iodepth=16 \
7
    --rw=randwrite --bs=4k --direct=1 --size=1G \
8
    --numjobs=1 --runtime=60 --group_reporting
9

10
# 顺序写入（使用 fio）
11
fio --name=seqwrite --ioengine=libaio --iodepth=16 \
12
    --rw=write --bs=1M --direct=1 --size=1G \
13
    --numjobs=1 --runtime=60 --group_reporting

1
/var/lib/kafka/data/orders-0/
2
├── 00000000000000000000.log      # 消息数据
3
├── 00000000000000000000.index    # 偏移量索引
4
├── 00000000000000000000.timeindex # 时间戳索引
5
├── 00000000000005367851.log
6
├── 00000000000005367851.index
7
├── 00000000000005367851.timeindex
8
└── leader-epoch-checkpoint       # Leader Epoch 检查点

1
// Log Segment 的核心结构
2
public class LogSegment {
3
    private final FileRecords log;          // .log 文件
4
    private final OffsetIndex offsetIndex;  // .index 文件
5
    private final TimeIndex timeIndex;      // .timeindex 文件
6
    private final long baseOffset;          // 段的起始 Offset
7

8
    // 追加消息
9
    public void append(long offset, MemoryRecords records) {
10
        // 1. 写入 .log 文件
11
        log.append(records);
12
        // 2. 更新偏移量索引（稀疏索引，每隔一定字节记录一条）
13
        if (shouldCreateIndexEntry()) {
14
            offsetIndex.append(offset, log.sizeInBytes());
15
        }
16
        // 3. 更新时间戳索引
17
        timeIndex.maybeAppend(timestamp, offset);
18
    }
19

20
    // 查找消息
21
    public FileRecords.FileChannelRecordSearchResult searchForOffset(
22
            long targetOffset, int startingPosition) {
23
        // 1. 二分查找 .index 定位到接近的物理位置
24
        // 2. 从该位置扫描 .log 找到精确的 Offset
25
        return offsetIndex.lookup(targetOffset);
26
    }
27
}

1
# 索引相关配置
2
# 索引间隔字节（越小越精确，但内存占用越大）
3
log.index.interval.bytes=4096
4

5
# 索引文件最大大小
6
log.index.size.max.bytes=10485760
7

8
# 段大小
9
log.segment.bytes=1073741824
10

11
# 段滚动时间（即使未满也滚动）
12
log.roll.hours=168
13

14
# 查看 Segment 详情
15
kafka-log-dirs --bootstrap-server localhost:9092 \
16
    --describe --topic-list orders

1
# Delete 策略配置
2
log.cleanup.policy=delete
3
log.retention.hours=168        # 保留 7 天
4
log.retention.bytes=1073741824 # 每个分区最大 1GB
5
log.retention.check.interval.ms=300000  # 每 5 分钟检查一次
6

7
# Compact 策略配置
8
log.cleanup.policy=compact
9
log.cleaner.min.compaction.lag.ms=0
10
log.cleaner.max.compaction.lag.ms=9223372036854775807
11
log.cleaner.dedupe.buffer.size=134217728
12
log.cleaner.threads=1
13
log.cleaner.backoff.ms=15000
14

15
# 同时使用两种策略
16
log.cleanup.policy=delete,compact

1
# 查看 Page Cache 使用情况
2
vmtouch /var/lib/kafka/data/orders-0/
3

4
# 输出示例：
5
#            Files: 6
6
#      Directories: 1
7
#    Resident Pages: 245760/262144  960M/1G  93.75%
8
#           Elapsed: 0.003432 seconds
9

10
# 手动释放 Page Cache（仅用于测试！）
11
echo 1 > /proc/sys/vm/drop_caches
12

13
# Kafka 相关的 OS 调优
14
# 增大文件描述符限制
15
ulimit -n 100000
16

17
# 调整脏页刷回策略
18
# 脏页占内存百分比达到此值时触发刷回
19
sysctl vm.dirty_background_ratio=5
20
# 脏页占内存百分比达到此值时阻塞写入
21
sysctl vm.dirty_ratio=80
22

23
# 调整 swappiness（尽量不用 swap）
24
sysctl vm.swappiness=1

1
// Kafka 的文件写入使用 FileChannel
2
// 数据写入 Page Cache，由 OS 决定何时刷盘
3
FileChannel channel = FileChannel.open(path,
4
    StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE);
5

6
// 写入数据（进入 Page Cache）
7
channel.write(buffer);
8

9
// 强制刷盘（fsync）—— Kafka 默认不主动调用
10
// 仅在特定配置下触发
11
channel.force(true);  // metadata = true 表示同时刷元数据
12

13
// Kafka 刷盘配置
14
// log.flush.interval.messages=10000  # 每 10000 条消息刷盘
15
// log.flush.interval.ms=1000        # 每 1000ms 刷盘
16
// 注意：通常不需要手动配置刷盘，依赖 OS 的刷回策略即可

1
# 查看预读设置
2
blockdev --getra /dev/sda1
3

4
# 设置预读大小（单位：512 字节扇区）
5
# 256 表示预读 128KB
6
blockdev --setra 256 /dev/sda1
7

8
# 对于 Kafka 消费场景，较大的预读值可以提升性能
9
# 推荐 256-4096（128KB-2MB）
10
blockdev --setra 4096 /dev/sda1

1
// Kafka 的零拷贝实现
2
// Kafka 使用 Java 的 FileChannel.transferTo() → 底层调用 sendfile()
3

4
// 消费者请求处理（简化版）
5
public class FetchResponse {
6
    public void writeTo(GatheringByteChannel channel) {
7
        // 零拷贝发送日志数据
8
        for (FileRecords fileRecords : records) {
9
            fileRecords.writeTo(channel, 0, fileRecords.sizeInBytes());
10
            // 底层调用：
11
            // fileChannel.transferTo(position, count, socketChannel)
12
            // → Linux sendfile() 系统调用
13
        }
14
    }
15
}
16

17
// FileChannel.transferTo 的零拷贝实现
18
// Linux 2.4+ 支持 scatter-gather DMA
19
// 数据从磁盘直接到网卡，不经过 CPU
20
FileChannel srcChannel = new FileInputStream(logFile).getChannel();
21
FileChannel destChannel = new FileOutputStream(socket).getChannel();
22
srcChannel.transferTo(0, srcChannel.size(), destChannel);

1
# 验证零拷贝是否生效
2
# 使用 strace 追踪系统调用
3
strace -e trace=sendfile -p $(pgrep -f kafka) 2>&1 | head -20
4

5
# 使用 perf 监控 sendfile 调用
6
perf stat -e 'syscalls:sys_enter_sendfile' -p $(pgrep -f kafka) sleep 10

1
// 生产者压缩配置
2
Properties props = new Properties();
3
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
4

5
// 批量发送配置（与压缩配合）
6
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);    // 16KB 批次
7
props.put(ProducerConfig.LINGER_MS_CONFIG, 5);          // 等待 5ms 凑批
8

9
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
10

11
// 压缩发生在 RecordBatch 层面
12
// 多条消息被压缩为一个 RecordBatch
13
// Broker 存储和传输的都是压缩后的数据
14
// 消费者收到后才解压
15

16
// 压缩效果对比
17
// 原始数据：1000 条 JSON 消息，约 500KB
18
// gzip：约 50KB（10:1）
19
// snappy：约 150KB（3.3:1）
20
// lz4：约 160KB（3.1:1）
21
// zstd：约 80KB（6.25:1）

1
// Segment 的生命周期管理
2
public class Log {
3
    private final ConcurrentMap<Long, LogSegment> segments;
4

5
    // 滚动到新 Segment
6
    public void roll(OptionalLong nextOffset) {
7
        // 1. 刷新当前 Segment
8
        if (activeSegment != null) {
9
            activeSegment.flush();
10
        }
11
        // 2. 创建新 Segment
12
        long newOffset = nextOffset.orElseGet(() ->
13
            activeSegment != null ? activeSegment.readNextOffset() : 0);
14
        LogSegment newSegment = LogSegment.open(dir, newOffset, config);
15
        // 3. 添加到 segments 集合
16
        segments.put(newOffset, newSegment);
17
    }
18

19
    // 删除旧 Segment
20
    public void deleteRetentionMsBreachedSegments() {
21
        // 1. 找到超过保留期的 Segment
22
        // 2. 从 segments 集合中移除
23
        // 3. 异步删除文件
24
    }
25
}

1
# Compact 清理过程
2
# 1. 选择最脏的 Segment（重复 Key 比例最高）
3
# 2. 将 Segment 分成 clean 和 dirty 两部分
4
# 3. 遍历 dirty 部分，保留每个 Key 的最新值
5
# 4. 将清理后的数据写入新 Segment
6
# 5. 替换旧 Segment
7

8
# Compact 相关配置
9
log.cleaner.enable=true
10
log.cleaner.threads=2
11
log.cleaner.dedupe.buffer.size=134217728    # 去重缓冲区 128MB
12
log.cleaner.io.buffer.load.factor=0.9
13
log.cleaner.backoff.ms=15000
14
log.cleaner.min.cleanable.dirty.ratio=0.5   # dirty 比例超过 50% 才清理

1
# XFS 文件系统创建和挂载
2
mkfs.xfs -f /dev/sdb1
3
mount -o noatime,nodiratime /dev/sdb1 /var/lib/kafka/data
4

5
# /etc/fstab 持久化
6
/dev/sdb1 /var/lib/kafka/data xfs noatime,nodiratime 0 2
7

8
# JBOD 配置（多磁盘）
9
# server.properties
10
log.dirs=/disk1/kafka,/disk2/kafka,/disk3/kafka,/disk4/kafka
11

12
# 磁盘空间监控
13
df -h /var/lib/kafka/data/
14
du -sh /var/lib/kafka/data/*/

1
# Broker 存储相关配置
2
# 段大小
3
log.segment.bytes=1073741824        # 1GB
4

5
# 刷盘策略（通常不需要修改）
6
# log.flush.interval.messages=10000
7
# log.flush.interval.ms=1000
8

9
# 保留策略
10
log.retention.hours=168             # 7 天
11
log.retention.bytes=-1              # 不限制大小
12
log.retention.check.interval.ms=300000
13

14
# 清理策略
15
log.cleanup.policy=delete
16
log.cleaner.enable=true
17

18
# 文件描述符
19
# /etc/security/limits.conf
20
# kafka soft nofile 100000
21
# kafka hard nofile 100000

1
// 监控磁盘使用
2
// JMX MBean: kafka.log:type=Log,name=Size,topic=*,partition=*
3
// JMX MBean: kafka.log:type=LogManager,name=LogFlushRateAndTimeMs
4

5
// 使用 kafka-log-dirs 查看磁盘使用
6
kafka-log-dirs --bootstrap-server localhost:9092 --describe