之前参加 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 来配置刷盘策略