嗨，你好呀，我是猿java

Log 是Kafka的核心组件之一，用于持久化存储消息，为了有效管理存储空间和保证系统性能，Kafka 提供了日志保留和数据清理策略。这篇文章，我将详细分析它们的工作原理。

日志保留期 {#日志保留期}

Kafka 的日志保留策略决定了消息在 Kafka 中存储的时间长度，保留策略可以基于时间或日志大小来配置。当消息超过指定的保留时间或日志大小限制时，Kafka 将自动清理这些消息以释放存储空间。

日志保留配置 {#日志保留配置}

Kafka 提供了多种配置选项以控制日志保留策略：

log.retention.hours: 定义消息在日志中保留的时间（以小时为单位），默认值为 168 小时（7 天）。
log.retention.minutes: 以分钟为单位的保留时间。
log.retention.ms: 以毫秒为单位的保留时间。
log.retention.bytes: 定义每个日志分区允许使用的最大存储空间，当达到此限制时，最早的消息将被删除。

需要注意的是，时间和大小限制是互斥的，Kafka 将依据首先满足的条件来清理日志。

日志清理策略 {#日志清理策略}

Kafka 提供两种主要的日志清理策略：

删除策略（delete）: 在达到保留期后删除旧数据。
压缩策略（compact）: 针对具有相同键的记录，只保留最新版本。

默认情况下，Kafka 使用删除策略。日志清理策略可以通过 log.cleanup.policy 配置，其中 delete 和 compact 都可以作为其值。

日志清理机制原理 {#日志清理机制原理}

Kafka 的日志清理是在后台运行的，它并不影响正常的读写操作，日志清理策略主要包含删除策略和压缩策略 2种类型：

删除策略 {#删除策略}

删除策略是最简单的日志清理机制，Kafka 定期检查日志分区的时间戳或大小，当某个分区超过指定的保留时间或大小时，系统会删除该分区的旧日志段（Log Segment）。具体过程如下：

检查条件: Kafka 定期比较当前时间与日志段创建时间的差值，或检查日志分区的大小是否超过配置的限制。
标记删除: 符合删除条件的日志段被标记为删除。
物理删除: 在下一个清理周期中，Kafka 将实际删除这些标记的日志段以释放磁盘空间。

压缩策略 {#压缩策略}

压缩策略主要用于仅保留每个键的最新消息版本，它适用于更新频繁的场景，例如数据库变更日志。压缩策略的工作流程如下：

收集日志段: Kafka 定期扫描日志段，识别出需要压缩的段。
构建索引: 为每个日志段构建一个映射，记录每个键的最新偏移量。
合并日志段: 确定每个键的最新消息后，Kafka 将这些消息写入新的日志段。
替换旧日志段: 新日志段生成后，Kafka 替换旧的日志段，并在下次清理时删除旧段。

核心源码分析 {#核心源码分析}

为了更深入理解 Kafka 的日志清理机制，接下来会分析几个相关的核心源码类：

LogCleaner 类 {#LogCleaner-类}

LogCleaner 是 Kafka 中负责日志压缩（compaction）的核心组件之一，它的主要功能是定期扫描 Kafka 日志，并对其进行压缩，以确保每个键只保留最新的值。下面是对 LogCleaner 源码的详细分析。

LogCleaner 的基本结构 {#LogCleaner-的基本结构}

LogCleaner 继承自 ShutdownableThread，这意味着它是一个可以安全关闭的后台线程，其主要职责是从需要压缩的日志中清除冗余消息。

|------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 | public class LogCleaner extends ShutdownableThread { // 主要成员变量 private final CleanerConfig config; private final OffsetCheckpoint checkpoint; private final Time time; private final Cleaner cleaner; public LogCleaner(String name, CleanerConfig config, OffsetCheckpoint checkpoint, Time time) { super(name, true); this.config = config; this.checkpoint = checkpoint; this.time = time; this.cleaner = new Cleaner(config, time); } @Override public void doWork() { // 核心清理逻辑 } } |

核心方法分析 {#核心方法分析}

doWork() {#doWork}

doWork() 是 LogCleaner 的核心方法，它被定期调用以执行日志压缩任务。

|---------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 | @Override public void doWork() { // 从清理队列中获取下一个需要清理的日志 LogToClean logToClean = cleanerManager.grabFilthiestLog(); if (logToClean != null) { try { // 执行压缩 cleaner.clean(logToClean); } finally { // 释放资源 cleanerManager.doneCleaning(logToClean); } } else { // 如果没有日志需要清理，则线程休眠一段时间 time.sleep(config.backOffMs); } } |

该方法的主要步骤包括：

从 cleanerManager 中获取下一个需要清理的日志。
调用 cleaner.clean() 方法对日志进行压缩。
完成后，释放资源并更新清理状态。

clean() {#clean}

clean() 方法是 Cleaner 类中的一个重要方法，负责具体的日志压缩操作。

|---------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 | public void clean(LogToClean logToClean) { // 获取需要压缩的日志段 List<LogSegment> segments = logToClean.segments(); // 创建一个新的日志段用于存储压缩后的数据 LogSegment newSegment = new LogSegment(...); // 遍历旧段，压缩数据并写入新段 for (LogSegment segment : segments) { // 读取每个消息 for (MessageAndOffset message : segment) { // 检查是否是最新的消息 if (isLatest(message)) { newSegment.append(message); } } } // 替换旧段 logToClean.replaceSegments(newSegment); } |

clean() 方法的主要步骤包括：

获取需要压缩的日志段。
创建新的日志段以存储压缩后的数据。
遍历旧日志段，选出每个键的最新消息并写入新段。
替换旧日志段为新段。

LogSegment 类 {#LogSegment-类}

LogSegment 是 Kafka 中表示日志文件的基本单位。每个 Kafka 主题分区由多个日志段（LogSegment）组成。每个日志段包括一个日志文件和一个索引文件。下面是对 LogSegment 类的源码分析，帮助理解其结构和功能。

LogSegment 的基本结构 {#LogSegment-的基本结构}

LogSegment 类位于 Kafka 的 log 包中，表示一个日志段。它包含两个主要文件：数据文件（存储消息）和索引文件（存储消息的偏移量）。

|---------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 6 7 8 9 10 11 | public class LogSegment { private final File log; private final FileMessageSet messageSet; private final OffsetIndex index; private final TimeIndex timeIndex; private final long baseOffset; private final long created; private final AtomicLong nextOffset; private final AtomicLong nextTimeIndexEntry; // 其他成员变量和方法 } |

核心构造函数 {#核心构造函数}

LogSegment 的构造函数负责初始化日志段的各个组件，包括数据文件和索引文件。

|---------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | public LogSegment(File logFile, FileMessageSet messageSet, OffsetIndex offsetIndex, TimeIndex timeIndex, long baseOffset, long created) { this.log = logFile; this.messageSet = messageSet; this.index = offsetIndex; this.timeIndex = timeIndex; this.baseOffset = baseOffset; this.created = created; this.nextOffset = new AtomicLong(baseOffset); this.nextTimeIndexEntry = new AtomicLong(baseOffset); } |

主要方法分析 {#主要方法分析}

append() {#append}

append() 方法用于向日志段追加消息，它将消息写入数据文件，并在索引文件中记录偏移量信息。

|---------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | public void append(long offset, RecordBatch batch) { // 将消息追加到数据文件 int physicalPosition = messageSet.append(batch); // 更新偏移量索引 index.append(offset, physicalPosition); // 更新时间索引 if (batch.maxTimestamp() > 0) { timeIndex.maybeAppend(batch.maxTimestamp(), offset); } // 更新下一个可用偏移量 nextOffset.set(offset + 1); } |

read() {#read}

read() 方法用于从日志段读取消息，它根据给定的偏移量和大小，返回相应的消息集合。

|-------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 | public FileMessageSet read(long startOffset, int maxSize) { // 计算读取的起始位置和大小 int startPosition = index.lookup(startOffset).position; return messageSet.read(startPosition, maxSize); } |

delete() {#delete}

delete() 方法用于删除日志段的物理文件，它会删除数据文件和索引文件。

|-------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | 1 2 3 4 5 6 7 8 | public void delete() { boolean deletedLog = log.delete(); boolean deletedIndex = index.delete(); boolean deletedTimeIndex = timeIndex.delete(); if (!deletedLog || !deletedIndex || !deletedTimeIndex) { throw new KafkaException("Failed to delete log segment files."); } } |