Kafka 如何保证消息的可靠性

public static void main(String[] args) {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "bigdata02:9092");
    props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer01");
    props.put(ProducerConfig.ACKS_CONFIG,"-1");
    //设置序列化的类
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    //初始化生产者
    KafkaProducer<Integer, String> producer = new KafkaProducer<>(props);
        ...
}

def appendRecords(timeout: Long,//请求处理的超时时间
                    requiredAcks: Short,//请求acks设置
                    internalTopicsAllowed: Boolean,//是否允许写入内部主题
                    isFromClient: Boolean,//写入方来源
                    entriesPerPartition: Map[TopicPartition, MemoryRecords],//待写入消息
                    responseCallback: Map[TopicPartition, PartitionResponse] => Unit,//回调逻辑
                    delayedProduceLock: Option[Lock] = None,//用来保护消费者组操作线程安全的锁对象
                   //消息格式转换操作的回调统计逻辑，主要用于统计消息格式转换操作过程中的一些数据指标，比如总共转换了多少条消息，花费多长时间等
                    recordConversionStatsCallback: Map[TopicPartition, RecordConversionStats] => Unit = _ => ()
                   ) 
                   ...

if (isValidRequiredAcks(requiredAcks)) {
  val sTime = time.milliseconds
  //TODO 数据追加到本地日志
  val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed,
    isFromClient = isFromClient, entriesPerPartition, requiredAcks)
  ...

//是否需要等待其它副本完成写入，通过delayedProduceRequestRequired方法来判断写入是否成功，
//如果为true，表示需要等待其它副本同步完成消息写入
if (delayedProduceRequestRequired(requiredAcks, entriesPerPartition, localProduceResults)) {
  val produceMetadata = ProduceMetadata(requiredAcks, produceStatus)
  //创建DelayedProduce延时请求对象
  val delayedProduce = new DelayedProduce(timeout, produceMetadata, this, responseCallback, delayedProduceLock)


  val producerRequestKeys = entriesPerPartition.keys.map(new TopicPartitionOperationKey(_)).toSeq
  //再一次尝试完成该延时请求
  //如果暂时无法完成，则将对象放入到相应的Purgatory中等待后续处理
  delayedProducePurgatory.tryCompleteElseWatch(delayedProduce, producerRequestKeys)

//是否需要手动刷盘。一般情况下不需要设置Broker端参数log.flush.interval.messages
//刷盘操作交由操作系统来完成。但某些情况下，可以设置该参数来确保高可靠性
//config.flushInterval默认值为Long.MaxValue，即默认这个分支条件不会满足，不执行flush方法
if (unflushedMessages >= config.flushInterval)
  flush()

def startup() {
  if (scheduler != null) {
    info("Starting log cleanup with a period of %d ms.".format(retentionCheckMs))
    //定时清理超时的文件
    //清理间隔：由参数 log.retention.check.interval.ms 配置，默认5分钟
    scheduler.schedule("kafka-log-retention",
                       cleanupLogs _,
                       delay = InitialTaskDelayMs,
                       period = retentionCheckMs,
                       TimeUnit.MILLISECONDS)
    info("Starting log flusher with a default period of %d ms.".format(flushCheckMs))
    //定时将内存中的数据刷写磁盘
    //时间间隔：由参数 log.flush.scheduler.interval.ms 配置，默认为Long.MAXVALUE
    scheduler.schedule("kafka-log-flusher",
                       flushDirtyLogs _,
                       delay = InitialTaskDelayMs,
                       period = flushCheckMs,
                       TimeUnit.MILLISECONDS)
    //定时更新检查点文件，kafka有时会重启，重启会依赖于检查点文件恢复数据
    //时间间隔：由参数 log.flush.offset.checkpoint.interval.ms 配置，默认60秒
    scheduler.schedule("kafka-recovery-point-checkpoint",
                       checkpointLogRecoveryOffsets _,
                       delay = InitialTaskDelayMs,
                       period = flushRecoveryOffsetCheckpointMs,
                       TimeUnit.MILLISECONDS)
    //定时更新Log起始偏移量检查点
    //时间间隔：由参数 log.flush.start.offset.checkpoint.interval.ms 配置，默认60秒
    scheduler.schedule("kafka-log-start-offset-checkpoint",
                       checkpointLogStartOffsets _,
                       delay = InitialTaskDelayMs,
                       period = flushStartOffsetCheckpointMs,
                       TimeUnit.MILLISECONDS)
    //清理待删除文件
    scheduler.schedule("kafka-delete-logs",
                       deleteLogs _,
                       delay = InitialTaskDelayMs,
                       unit = TimeUnit.MILLISECONDS)
  }
  //默认为true
  if (cleanerConfig.enableCleaner)
    cleaner.startup()
}