配置 kafka 同步刷盘

Posted 2021-03-01 allenwas3

之前参加 rocketmq 的 meetup，台上有人讲，kafka 不支持同步刷盘，当时没太在意，今天抽空看了下代码：

kafka 提供了配置参数来支持同步刷盘，和 rocktmq 的做法不同（4.7 的 rmq 在 sync_disk 模式，统一在 GroupCommitService 中刷盘，不阻塞 SendMessageProcessor 线程）

kafka.log.Log#append

看下代码片段就好，不深入讲

// always update the last producer id map offset so that the snapshot reflects the current offset
// even if there isn‘t any idempotent data being written
producerStateManager.updateMapEndOffset(appendInfo.lastOffset + 1)

// increment the log end offset
updateLogEndOffset(appendInfo.lastOffset + 1)

// update the first unstable offset (which is used to compute LSO)
updateFirstUnstableOffset()

trace(s"Appended message set with last offset: ${appendInfo.lastOffset}, " +
  s"first offset: ${appendInfo.firstOffset}, " +
  s"next offset: ${nextOffsetMetadata.messageOffset}, " +
  s"and messages: $validRecords")

// 我们所需要的同步刷盘
if (unflushedMessages >= config.flushInterval)
  flush()

appendInfo

再看下文档

public static final String FLUSH_MESSAGES_INTERVAL_CONFIG = "flush.messages";
public static final String FLUSH_MESSAGES_INTERVAL_DOC = "This setting allows specifying an interval at " +
    "which we will force an fsync of data written to the log. For example if this was set to 1 " +
    "we would fsync after every message; if it were 5 we would fsync after every five messages. " +
    "In general we recommend you not set this and use replication for durability and allow the " +
    "operating system‘s background flush capabilities as it is more efficient. This setting can " +
    "be overridden on a per-topic basis (see <a href="#topicconfigs">the per-topic configuration section</a>).";

public static final String FLUSH_MS_CONFIG = "flush.ms";
public static final String FLUSH_MS_DOC = "This setting allows specifying a time interval at which we will " +
    "force an fsync of data written to the log. For example if this was set to 1000 " +
    "we would fsync after 1000 ms had passed. In general we recommend you not set " +
    "this and use replication for durability and allow the operating system‘s background " +
    "flush capabilities as it is more efficient.";

所以，可以通过 flush.messages 和 flush.ms 来配置刷盘策略