日志存储与清理策略

Kafka 的一切都建立在"追加写日志文件"上：消息怎么存、什么时候滚段、怎么删、怎么压缩，决定了 retention 配置到底怎么生效、为什么改历史配置"不立即生效"、以及 cleanup.policy=compact 为什么能把 Topic 当数据库用。理解存储层，运维决策才有依据。

前置阅读：00-安装与配置（log.dirs/log.retention.* 参数）、01-核心概念（顺序写/零拷贝）。

分区目录与 Segment

一个 2 分区的 Topic order-events 落在 log.dirs 下就是两个目录：

kafka-logs/
└── order-events-0/                    # 分区目录：<topic>-<partition>
    ├── 00000000000000000000.log       # 数据段：消息实体（追加写）
    ├── 00000000000000000000.index     # 偏移量索引（稀疏）
    ├── 00000000000000000000.timeindex # 时间戳索引（稀疏）
    ├── 000000000000011223344.log      # 上一段已滚走，等待过期清理
    ├── ...
    ├── leader-epoch-checkpoint        # Leader 纪元记录（截断/恢复用）
    ├── snapshot                         # 分区元数据快照（恢复加速）
    └── .lock                          # 目录锁（同机多 Broker 抢目录的防御）

• 文件名是该段第一条消息的 Offset，左补零到 21 位；段按 Offset 天然有序，定位一条消息 = 找段 + 段内二分。
• 活动段（最后一段）只有 .log 在被追加；写满 log.segment.bytes（默认 1GB）或超过 log.segment.ms 就滚出下一段。滚段粒度直接影响吞吐：段太小 → 文件句柄/索引开销大；太大 → 单段不可变清理不灵活。生产上一般 256MB~1GB，很少需要动默认值。

稀疏索引：为什么查 Offset 是 O(log N)

.index 不是"每条消息一个索引项"，而是每隔一段字节记录一次（log.index.interval.bytes 默认 4096）：

偏移量索引条目： offset → 文件物理位置
11223301  → 0
11223342  → 4096      ← 这中间的消息没有索引项
11223388  → 8192

查询 offset=11223350：对索引二分找到"不大于目标的最大项"（11223342→4096），再从该物理位置顺序扫描直到目标。索引因此极小（万级条目/段），能整文件 mmap 进内存——这就是"Kafka 随机读也快"的底气。.timeindex 同理服务按时间查询：先时间→Offset，再走上面的路。kafka-get-offsets.bat --time（02 章）和 franz 的 kgo.NewOffset().AtStart()/AtEnd()/At(n)/AfterMilli(...) 起点配方、03 章 scan 客户端的非组定位，底下都是这套。

注意：稀疏索引意味着"按 Offset 点查"和"按时间回放"都只能定位到段附近再扫，Kafka 不是数据库——不要设计"按任意业务字段查单条消息"的功能，那是 ES/ClickHouse 的活。

删除策略：cleanup.policy=delete（默认）

retention 判定以段为单位，不是以消息为单位：

1. Broker 每个 log.retention.check.interval.ms（默认 5 分钟）扫一遍非活动段；
2. 段的最大时间戳超过 retention.ms（或分区超出 log.retention.bytes）→ 整段删除；
3. 活动段永不被删——哪怕它已经"过期"。

由此推出三个真实行为：

• “保留 7 天”≠ 精确 7 天：一条消息在 6 天 23 小时进的一个大段里，它实际会活到"整段过期 + 检查周期到点"，可能是 8 天；反之段滚得勤就删得快。
• 改了 retention.ms 不会追溯已删数据，也不会立刻删存量：它只影响下一轮段级判定。
• 时间戳类型影响判定基准：Topic 是 LogAppendTime（04 章）时段内时间单调，删除干净利落；CreateTime（默认）下生产者乱序塞入老时间戳消息会让段"提前过期"被整段删掉——对时间敏感的 Topic 考虑用 message.timestamp.type=LogAppendTime。

# Topic 级覆盖：业务日志 3 天、审计 Topic 30 天
.\bin\windows\kafka-configs.bat --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name app-logs --alter --add-config retention.ms=259200000

主动删数据（合规"被遗忘权"）：kafka-delete-records 可以给分区打"下界标记"，让早于某 Offset/时间戳的数据尽快物理删除：

.\bin\windows\kafka-delete-records.bat --bootstrap-server localhost:9092 \
  --offset-json-file reset.json   # json: {"partitions":[{"topic":"users","partition":0,"offset":1000}]}

它删的是"已滚出的完整段"，活动段里的内容仍要等滚段——对强合规场景，配合 segment.ms 调小才有实际意义。

压缩策略：cleanup.policy=compact，把 Topic 当表用

compact 的哲学相反：不删时间，删"旧版本"——每个 Key 至少保留最新值。段滚出后压缩线程重写它：同 Key 只留最后一条；Value 为 null 的消息（墓碑）标记该 Key 已删除，墓碑本身在 delete.retention.ms（默认 24h）后被清掉。

# 建一个"用户状态表" Topic：CDC 每变更投一条同 Key 消息
.\bin\windows\kafka-topics.bat --create --topic user-profile \
  --bootstrap-server localhost:9092 --partitions 6 --replication-factor 3 \
  --config cleanup.policy=compact --config min.cleanable.dirty.ratio=0.3

触发时机由两个比例参数控制：脏数据（被覆盖的旧版本）占比达到 min.cleanable.dirty.ratio（默认 0.5）或超过 segment.ms/min.compaction.lag.ms 才动手。刚写入的段有 min.compaction.lag.ms 保护期，给慢消费者留出完整读历史的时间。

典型用法：新消费者先"建表"再"追增量"

用户画像场景——user-profile 每个 Key 一条最新状态，服务重启后不需要外部数据库，直接从 Topic 重建本地缓存：

// 非组模式（即 03 章定义的 scan 客户端）：ConsumePartitions 里写 AtStart，
// 全量扫压缩后的 Topic，构建最新状态视图。每个分区各建一份视图片段，或全分区各起一个客户端
scan, _ := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.ConsumePartitions(map[string]map[int32]kgo.Offset{
        "user-profile": {0: kgo.NewOffset().AtStart()},
    }),
)

type Profile struct{ Name, City string }
users := map[string]Profile{} // key → 最新状态

for {
    fs := scan.PollFetches(ctx)
    if err := fs.Err0(); errors.Is(err, context.Canceled) || errors.Is(err, kgo.ErrClientClosed) {
        break
    }
    fs.EachRecord(func(m *kgo.Record) {
        if m.Value == nil { // 墓碑：该 Key 已删除
            delete(users, string(m.Key))
            return
        }
        var p Profile
        json.Unmarshal(m.Value, &p)
        users[string(m.Key)] = p // 后写覆盖先写
    })
}

kafka-go 存量写法：kafka.NewReader(kafka.ReaderConfig{Topic: "user-profile", Partition: 0, StartOffset: kafka.FirstOffset}) + FetchMessage，循环形状同上；注意 StartOffset 零值也是 FirstOffset，这里反而"歪打正着"。

Kafka 自己的 __consumer_offsets 就是 compact Topic（50 分区、offsets.topic.replication.factor 控制副本）——你消费进度能被记住，靠的正是这个机制。

注意：compact 不保证"每个 Key 只剩一条"——同一 Key 的多个版本可以合法共存（压缩是异步渐进的），所以读 compact Topic 的代码必须自己"后写覆盖先写"（如上）。把它当"保证最终含最新值的日志"，而不是"表"。

另外 cleanup.policy=compact,delete 可以组合：先压缩再按时间删，适合"状态表 + 保留窗口"要兼得的场景。

消息格式与压缩：为什么端到端只压一次

v2 消息格式（0.11+，Kafka 4.x 已无老格式）里，.log 的最小单元不是消息而是 RecordBatch：一批消息共享压缩头、相对 Offset、相对时间戳、CRC，外加事务标记位。这决定了压缩的真实形态：

• 生产端批量压缩：kafka-go Writer 把攒好的批（BatchSize/BatchTimeout，04 章）整体压缩后再发；
• Broker 端原样落盘：默认不解压（只有改格式/建事务标记才解开重压）——网络传输省一次、磁盘存储省一次，都是生产者选的编码；
• 消费端解压：Reader 拉回整批在本地解压，CPU 花在读端。

编码选择（Compression: kafka.Gzip / Snappy / Lz4 / Zstd，v0.4 起内置无需额外 import）的经验排序：Zstd 默认首选（同等 CPU 下压缩率最好，且 Kafka 3.x 起 broker 内部 Topic 也用它）；延迟敏感选 Lz4/Snappy；Gzip 兼容性最好但 CPU 高，基本只在跨老版本兜底时用。压缩与批是乘法关系：批越大 → 同前缀模式重复越多 → 压缩率越高，这也是 BatchTimeout 敢设到几十毫秒不心疼延迟的原因。

零拷贝链路补一刀：消费者按 FetchOffset + MaxBytes 拉的是"段文件的一段字节"，sendfile() 直接把 page cache 的数据递进 socket——压缩过的批对 Broker 是透明字节，Broker 根本不看内容。存储格式与网络路径在这里合流，这就是 01 章"Kafka 为什么快"的存储侧拼图。

分层存储（Tiered Storage）：远期的保留成本解法

本地磁盘贵，但业务想要 30 天回放 + 历史全量重算。KIP-405 分层存储的方向：segment 滚出后异步复制到对象存储（S3/OSS），本地只留热段（local.retention.ms），冷数据按 Offset 从远端取。它不改变消费者语义（retention 从"本地磁盘时长"变成"本地+远端总时长"），但大幅降低单位保留成本。

现状（2026）：自 3.6 起 long-term preview、4.x 持续打磨（4.3 修了若干远端复制边界问题），仍是 Early Access，生产核心集群慎开；云厂商托管版（如 AWS MSK 的 Tiered Storage）已商用，走云可以提前受益。选型时先问自己：数据真要留那么久吗？大多数"要留很久"的需求，答案其实是 compact 或数据湖。

实战建议

1. retention 按"重放窗口"设计：消费者出 bug 要跑多久才能追回来？留 retention.ms ≥ 追数时长 × 3。默认 7 天对日志采集合适，对需要全量重算的流处理偏短。
2. 段大小 × 副本数 × 分区数 = 你的文件句柄账：每分区每副本一组 .log/.index/.timeindex，动 log.segment.bytes 和分区数之前先查 ulimit -n。
3. compact Topic 三件套：cleanup.policy=compact + 业务 Key 稳定 + delete.retention.ms ≥ 最慢消费者窗口；没有这三条，“表"就会读出幻觉数据。
4. 磁盘用 JBOD 直通（多块盘各挂一个 log.dirs 路径，Kafka 自己分摊分区），别做 RAID5——顺序写放大和恢复惩罚在 RAID 里全吃亏，云盘同理选本地 NVMe 或高 IOPS 云盘按目录拆分。
5. 存储类参数（log.segment.bytes、flush.*、index.interval.bytes）属于"默认值即专家值”——除非你已从 metrics 定位到具体瓶颈，否则不要动，动之前先读一遍 Kafka Improvement Proposal 原文里的假设前提。

参考资料

• Kafka 文档 - Design（日志与索引）
• KIP-405: Tiered Storage