深度剖析 Kafka 日志保留與數據清理策略

作者：猿java 2024-10-23 16:06:50

開發

本文從原理到源碼詳細分析了 Kafka 的日志保留和數據清理策略，Kafka 能夠在保證數據持久化的同時，最大限度地利用存儲資源。。

Log 是Kafka的核心組件之一，用于持久化存儲消息，為了有效管理存儲空間和保證系統性能，Kafka 提供了日志保留和數據清理策略。這篇文章，我將詳細分析它們的工作原理。

一、日志保留

Kafka 的日志保留策略決定了消息在 Kafka 中存儲的時間長度，保留策略可以基于時間或日志大小來配置。當消息超過指定的保留時間或日志大小限制時，Kafka 將自動清理這些消息以釋放存儲空間。

1.日志保留配置

Kafka 提供了多種配置選項以控制日志保留策略：

log.retention.hours: 定義消息在日志中保留的時間（以小時為單位），默認值為 168 小時（7 天）。
log.retention.minutes: 以分鐘為單位的保留時間。
log.retention.ms: 以毫秒為單位的保留時間。
log.retention.bytes: 定義每個日志分區允許使用的最大存儲空間，當達到此限制時，最早的消息將被刪除。

需要注意的是，時間和大小限制是互斥的，Kafka 將依據首先滿足的條件來清理日志。

2.日志清理策略

Kafka 提供兩種主要的日志清理策略：

刪除策略（delete）: 在達到保留期后刪除舊數據。
壓縮策略（compact）: 針對具有相同鍵的記錄，只保留最新版本。

默認情況下，Kafka 使用刪除策略。日志清理策略可以通過 log.cleanup.policy 配置，其中 delete 和 compact 都可以作為其值。

二、日志清理

Kafka 的日志清理是在后臺運行的，它并不影響正常的讀寫操作，日志清理策略主要包含刪除策略和壓縮策略 2種類型：

1.刪除策略

刪除策略是最簡單的日志清理機制，Kafka 定期檢查日志分區的時間戳或大小，當某個分區超過指定的保留時間或大小時，系統會刪除該分區的舊日志段（Log Segment）。具體過程如下：

檢查條件: Kafka 定期比較當前時間與日志段創建時間的差值，或檢查日志分區的大小是否超過配置的限制。
標記刪除: 符合刪除條件的日志段被標記為刪除。
物理刪除: 在下一個清理周期中，Kafka 將實際刪除這些標記的日志段以釋放磁盤空間。

2.壓縮策略

壓縮策略主要用于僅保留每個鍵的最新消息版本，它適用于更新頻繁的場景，例如數據庫變更日志。壓縮策略的工作流程如下：

收集日志段: Kafka 定期掃描日志段，識別出需要壓縮的段。
構建索引: 為每個日志段構建一個映射，記錄每個鍵的最新偏移量。
合并日志段: 確定每個鍵的最新消息后，Kafka 將這些消息寫入新的日志段。
替換舊日志段: 新日志段生成后，Kafka 替換舊的日志段，并在下次清理時刪除舊段。

三、核心源碼分析

為了更深入理解 Kafka 的日志清理機制，接下來會分析幾個相關的核心源碼類：

1.LogCleaner 類

LogCleaner 是 Kafka 中負責日志壓縮（compaction）的核心組件之一，它的主要功能是定期掃描 Kafka 日志，并對其進行壓縮，以確保每個鍵只保留最新的值。下面是對 LogCleaner 源碼的詳細分析。

(1) LogCleaner 的基本結構

LogCleaner 繼承自 ShutdownableThread，這意味著它是一個可以安全關閉的后臺線程，其主要職責是從需要壓縮的日志中清除冗余消息。

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() {
        // 核心清理邏輯
    }

(2) 核心方法分析

① doWork()

doWork() 是 LogCleaner 的核心方法，它被定期調用以執行日志壓縮任務。

@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() 方法是 Cleaner 類中的一個重要方法，負責具體的日志壓縮操作。

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() 方法的主要步驟包括：

獲取需要壓縮的日志段。
創建新的日志段以存儲壓縮后的數據。
遍歷舊日志段，選出每個鍵的最新消息并寫入新段。
替換舊日志段為新段。

2.LogSegment 類

LogSegment 是 Kafka 中表示日志文件的基本單位。每個 Kafka 主題分區由多個日志段（LogSegment）組成。每個日志段包括一個日志文件和一個索引文件。下面是對 LogSegment 類的源碼分析，幫助理解其結構和功能。

(1) LogSegment 的基本結構

LogSegment 類位于 Kafka 的 log 包中，表示一個日志段。它包含兩個主要文件：數據文件（存儲消息）和索引文件（存儲消息的偏移量）。

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;
    // 其他成員變量和方法
}

(2) 核心構造函數

LogSegment 的構造函數負責初始化日志段的各個組件，包括數據文件和索引文件。

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);
}

(3) 主要方法分析

① append()

append() 方法用于向日志段追加消息，它將消息寫入數據文件，并在索引文件中記錄偏移量信息。

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() 方法用于從日志段讀取消息，它根據給定的偏移量和大小，返回相應的消息集合。

public FileMessageSet read(long startOffset, int maxSize) {
    // 計算讀取的起始位置和大小
    int startPosition = index.lookup(startOffset).position;
    return messageSet.read(startPosition, maxSize);
}

③ delete()

delete() 方法用于刪除日志段的物理文件，它會刪除數據文件和索引文件。

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.");
    }
}

四、優化建議

Kafka 的日志清理機制可以通過多種配置進行優化，以適應不同的業務需求。以下是一些常見的優化建議：

合理設置保留時間：根據數據的重要性和訪問頻率，合理設置日志的保留時間。對于不常訪問的數據，可以適當縮短保留時間，以節省存儲空間。
調整日志段大小：通過設置 log.segment.bytes，可以控制每個日志段的大小。適當的日志段大小可以提高清理效率，避免頻繁的段切換。
配置清理線程：Kafka 允許配置清理線程的數量和頻率。通過 log.cleaner.threads 和 log.cleaner.interval.ms 配置，可以優化清理線程的性能。

五、總結

本文，我們從原理到源碼詳細分析了 Kafka 的日志保留和數據清理策略，在日常工作種，通過合理配置和優化這些策略，Kafka 能夠在保證數據持久化的同時，最大限度地利用存儲資源。

責任編輯：趙寧寧來源：猿java

Kafka 日志