阿里 P7 三面凉凉，kafka Borker 日志持久化没答上来

2023-12-18 15:01
文章标签 阿里 日志 持久 kafka 三面 p7 上来 凉凉 borker 化没答

本文主要是介绍阿里 P7 三面凉凉，kafka Borker 日志持久化没答上来，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

  • 👏作者简介：大家好，我是爱敲代码的小黄，阿里巴巴淘天Java开发工程师，CSDN博客专家
  • 📕系列专栏：Spring源码、Netty源码、Kafka源码、JUC源码、dubbo源码系列
  • 🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
  • 🍂博主正在努力完成2023计划中：以梦为马，扬帆起航，2023追梦人
  • 📝联系方式：hls1793929520，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀

在这里插入图片描述

文章目录

  • 阿里 P7 三面凉凉，kafka Borker 日志持久化没答上来
  • 一、引言
  • 二、日志原理介绍
  • 二、日志源码
    • 1、授权校验
    • 2、消息添加
      • 2.1 获取 Partition
      • 2.2 向 Leader 追加日志
        • 2.2.1 是否创建 segment
        • 2.2.2 创建 segment
          • 2.2.2.1 文件路径校验
          • 2.2.2.2 segment 参数
          • 2.2.2.3 生成 segment
        • 2.2.3 向 segment 添加日志
          • 2.2.3.1 稀疏索引
          • 2.2.3.2 偏移量索引
          • 2.2.3.3 时间戳索引
          • 2.2.3.4 索引总结
        • 2.2.4 flush刷新
      • 2.3 Follow 获取日志
    • 三、流程图
    • 四、总结

Kafka从成神到成仙系列

  • 【Kafka从成神到升仙系列 一】Kafka源码环境搭建
  • 【Kafka从成神到升仙系列 二】生产者如何将消息放入到内存缓冲区
  • 【Kafka从成神到升仙系列 三】你真的了解 Kafka 的元数据嘛
  • 【Kafka从成神到升仙系列 四】你真的了解 Kafka 的缓存池机制嘛
  • 【Kafka从成神到升仙系列 五】面试官问我 Kafka 生产者的网络架构，我直接开始从源码背起…
  • 【Kafka从成神到升仙系列 六】kafka 不能失去网络通信，就像西方不能失去耶路撒冷

阿里 P7 三面凉凉，kafka Borker 日志持久化没答上来

一、引言

前段时间有个朋友，去面了阿里集团的P7岗位，很遗憾的是三面没有过

其中有一个 kafka Borker 日志如何持久化的问题没有答上来

今天正好写一篇源码文章给朋友复盘一下

虽然现在是互联网寒冬，但乾坤未定，你我皆是黑马！

废话不多说，发车！

二、日志原理介绍

在讲 Kafka 日志源码之前，我们要先对 Kafka 日志有一个大体的认识

这也是阅读源码的关键，一步一步来

前面我们聊到了 Kafka 的生产端的整体架构

image.png

可以看到，我们每一个 Topic 都可以分为多个 Partition ，而每一个 Partition 对应着一个 Log

但这里会存在两个问题，如果我们的数据过大

  • 一个 Log 能装下吗？
  • 就算能装下，插入/查询速度怎么保证？

所以，Kafka 在这里引入了日志分段（LogSegment ）的概念，将一个 Log 切割成多个 LogSegment 进行存储

实际上，这里的 Log 和 LogSegment 并不是纯粹的物理意义上的概念

  • Log 对应的文件夹
  • LogSegment 对应磁盘上的一个日志文件和两个索引文件
    • 日志文件：以 .log 为文件后缀
    • 两个索引文件：
      • 偏移量索引文件（以 .index为文件后缀）
      • 时间戳索引文件（以.timeindex为文件后缀）

这里有个重点要记一下：每个 LogSegment 都有一个基准偏移量 baseOffset，用来表示当前 LogSegment 第一条消息的 offset

日志和索引文件命名都是按照基准偏移量进行命名，所以日志整体架构如下：

image.png

这里我们简单介绍下这个日志是怎么搜索的，后面会深入源码细聊

二、日志源码

我们回顾一下上篇文章的整体流程图：

在这里插入图片描述

我们可以看到，消息的处理是通过 KafkaApis 来进行的，日志持久化通过 case ApiKeys.PRODUCE => handleProduceRequest(request)

本篇我们也围绕这个方法展开

1、授权校验

def handleProduceRequest(request: RequestChannel.Request) {// authorizedRequestInfo：存储通过授权验证的主题分区和对应的内存记录。val authorizedRequestInfo = mutable.Map[TopicPartition, MemoryRecords]()for ((topicPartition, memoryRecords) <- produceRequest.partitionRecordsOrFail.asScala) {if (!authorize(request.session, Write, Resource(Topic, topicPartition.topic, LITERAL)))// 未授权的unauthorizedTopicResponses += topicPartition -> new PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED)else if (!metadataCache.contains(topicPartition))nonExistingTopicResponses += topicPartition -> new PartitionResponse(Errors.UNKNOWN_TOPIC_OR_PARTITION)elsetry {// 授权的ProduceRequest.validateRecords(request.header.apiVersion(), memoryRecords)authorizedRequestInfo += (topicPartition -> memoryRecords)} catch {case e: ApiException =>invalidRequestResponses += topicPartition -> new PartitionResponse(Errors.forException(e))}}
}

2、消息添加

  • 【重点】timeout：超时时间
  • 【重点】requiredAcks：指定了在记录追加到副本后需要多少个副本进行确认，才认为写操作成功
    • 0: 不需要任何副本的确认
    • 1: 只需要主副本确认
    • -1 或 all: 需要所有副本的确认
  • internalTopicsAllowed：是否允许将记录追加到内部主题
  • isFromClient：请求是否来自客户端
  • 【重点】entriesPerPartition：包含了通过授权验证的主题分区和对应的内存记录
  • responseCallback：回调函数，在记录追加完成后，会调用该回调函数发送响应给客户端。
  • recordConversionStatsCallback：处理记录转换统计信息的逻辑
replicaManager.appendRecords(timeout = produceRequest.timeout.toLong,requiredAcks = produceRequest.acks,internalTopicsAllowed = internalTopicsAllowed,isFromClient = true,entriesPerPartition = authorizedRequestInfo,responseCallback = sendResponseCallback,recordConversionStatsCallback = processingStatsCallback)

我们主要关心这三个参数即可：timeout、requiredAcks、entriesPerPartition，其余的目前不太重要

def appendRecords(timeout: Long,requiredAcks: Short,internalTopicsAllowed: Boolean,isFromClient: Boolean,entriesPerPartition: Map[TopicPartition, MemoryRecords],responseCallback: Map[TopicPartition, PartitionResponse] => Unit,delayedProduceLock: Option[Lock] = None,recordConversionStatsCallback: Map[TopicPartition, RecordConversionStats] => Unit = _ => ()) {// 校验当前的ACKif (isValidRequiredAcks(requiredAcks)) {// 记录起始时间val sTime = time.milliseconds// 追加本地日志val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed,isFromClient = isFromClient, entriesPerPartition, requiredAcks)}
}// 允许当前的ACK为1、0、-1
private def isValidRequiredAcks(requiredAcks: Short): Boolean = {requiredAcks == -1 || requiredAcks == 1 || requiredAcks == 0
}

这里的追加本地日志就是我们本篇的重点

2.1 获取 Partition

private def appendToLocalLog(internalTopicsAllowed: Boolean,isFromClient: Boolean,entriesPerPartition: Map[TopicPartition, MemoryRecords],requiredAcks: Short): Map[TopicPartition, LogAppendResult] = {val partition = getPartitionOrException(topicPartition, expectLeader = true)
}// 根据给定的主题分区获取对应的分区对象
def getPartitionOrException(topicPartition: TopicPartition, expectLeader: Boolean): Partition = {// 获取Partition并匹配getPartition(topicPartition) match {case Some(partition) =>if (partition eq ReplicaManager.OfflinePartition)throw new KafkaStorageException()elsepartitioncase None if metadataCache.contains(topicPartition) =>if (expectLeader) {throw new NotLeaderForPartitionException()} else {throw new ReplicaNotAvailableException()}}}

2.2 向 Leader 追加日志

val info = partition.appendRecordsToLeader(records, isFromClient, requiredAcks);def appendRecordsToLeader(records: MemoryRecords, isFromClient: Boolean, requiredAcks: Int = 0): LogAppendInfo = {val info = log.appendAsLeader(records, leaderEpoch = this.leaderEpoch, isFromClient,interBrokerProtocolVersion)
}def appendAsLeader(records: MemoryRecords, leaderEpoch: Int, isFromClient: Boolean = true,interBrokerProtocolVersion: ApiVersion = ApiVersion.latestVersion): LogAppendInfo = {append(records, isFromClient, interBrokerProtocolVersion, assignOffsets = true, leaderEpoch)}
2.2.1 是否创建 segment

这里就到了我们一开始图中的 LogSegment

 val segment = maybeRoll(validRecords.sizeInBytes, appendInfo);private def maybeRoll(messagesSize: Int, appendInfo: LogAppendInfo): LogSegment = {// 如果应该滚动，创建一个新的segment// 反之，则返回当前的segmentif (segment.shouldRoll(RollParams(config, appendInfo, messagesSize, now))) {appendInfo.firstOffset match {case Some(firstOffset) => roll(Some(firstOffset))case None => roll(Some(maxOffsetInMessages - Integer.MAX_VALUE))}} else {segment}}

一共有六个条件，触发这六个条件，就会重新创建一个 segment

  • timeWaitedForRoll(rollParams.now, rollParams.maxTimestampInMessages) > rollParams.maxSegmentMs - rollJitterMs ：判断时间等待是不是超时
  • size > rollParams.maxSegmentBytes - rollParams.messagesSize：当前 segment 是否有充足的空间存储当前信息
  • size > 0 && reachedRollMs ：当前日志段的大小大于0，并且达到了进行日志分段的时间条件reachedRollMs
  • offsetIndex.isFull ：偏移索引满了
  • timeIndex.isFull：时间戳索引满了
  • !canConvertToRelativeOffset(rollParams.maxOffsetInMessages)：无法进行相对偏移的转换操作
class LogSegment private[log] (val log: FileRecords,val offsetIndex: OffsetIndex,val timeIndex: TimeIndex,val txnIndex: TransactionIndex,val baseOffset: Long,val indexIntervalBytes: Int,val rollJitterMs: Long,val time: Time) extends Logging {def shouldRoll(rollParams: RollParams): Boolean = {val reachedRollMs = timeWaitedForRoll(rollParams.now, rollParams.maxTimestampInMessages) >    rollParams.maxSegmentMs - rollJitterMssize > rollParams.maxSegmentBytes - rollParams.messagesSize ||(size > 0 && reachedRollMs) ||offsetIndex.isFull || timeIndex.isFull || !canConvertToRelativeOffset(rollParams.maxOffsetInMessages)
}

整体来看，六个条件也比较简单，我们继续往后看

2.2.2 创建 segment
appendInfo.firstOffset match {// 存在第一个偏移量case Some(firstOffset) => roll(Some(firstOffset))// 不存在第一个偏移量case None => roll(Some(maxOffsetInMessages - Integer.MAX_VALUE))
}
2.2.2.1 文件路径校验
def roll(expectedNextOffset: Option[Long] = None): LogSegment = {// 获取最新的offsetval newOffset = math.max(expectedNextOffset.getOrElse(0L), logEndOffset)// 获取日志文件路径val logFile = Log.logFile(dir, newOffset)// 获取偏移量索引文件路径val offsetIdxFile = offsetIndexFile(dir, newOffset)// 获取时间戳索引文件路径val timeIdxFile = timeIndexFile(dir, newOffset)// 获取事务索引文件路径val txnIdxFile = transactionIndexFile(dir, newOffset)// 对路径列表进行遍历，如果文件存在，则将其删除。for (file <- List(logFile, offsetIdxFile, timeIdxFile, txnIdxFile) if file.exists) {Files.delete(file.toPath)}
}
2.2.2.2 segment 参数
  • dir：日志段所在的目录
  • baseOffset：日志段的基准偏移量
  • config：日志的配置信息
  • time：时间对象，用于处理时间相关的操作。
  • fileAlreadyExists：指示日志文件是否已经存在
  • initFileSize：初始文件大小
  • preallocate：是否预分配文件空间
  • fileSuffix：文件后缀
val segment = LogSegment.open(dir,baseOffset = newOffset,config,time = time,fileAlreadyExists = false,initFileSize = initFileSize,preallocate = config.preallocate)
2.2.2.3 生成 segment
new LogSegment(// 生成日志文件FileRecords.open(Log.logFile(dir, baseOffset, fileSuffix), fileAlreadyExists, initFileSize, preallocate),// 生成偏移量索引new OffsetIndex(Log.offsetIndexFile(dir, baseOffset, fileSuffix), baseOffset = baseOffset, maxIndexSize = maxIndexSize),// 生成时间戳索引new TimeIndex(Log.timeIndexFile(dir, baseOffset, fileSuffix), baseOffset = baseOffset, maxIndexSize = maxIndexSize),// 生成事务索引new TransactionIndex(baseOffset, Log.transactionIndexFile(dir, baseOffset, fileSuffix)),// 基准偏移量baseOffset,indexIntervalBytes = config.indexInterval,rollJitterMs = config.randomSegmentJitter,time)

这里有一个重点需要关注一下，那就是 mmap 的零拷贝

OffsetIndex 和 TimeIndex 他们继承 AbstractIndex ，而 AbstractIndex 中使用 mmp 作为 buffer 

class OffsetIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writable: Boolean = true) extends AbstractIndex[Long, Int](_file, baseOffset, maxIndexSize, writable) abstract class AbstractIndex{protected var mmap: MappedByteBuffer = {};
}

另外，这里先提一个知识点，后面会专门写一篇文章来分析一下

我们索引在查询的时候，采用的是二分查找的方式，这会导致 缺页中断，于是 kafka 将二分查找进行改进，将索引区分为 冷区 和 热区，分别搜索，尽可能保证热区的页在 Page Cache 里面，从而避免缺页中断。

当我们的 segment 生成完之后，就返回了

2.2.3 向 segment 添加日志
segment.append(largestOffset = appendInfo.lastOffset,largestTimestamp = appendInfo.maxTimestamp,shallowOffsetOfMaxTimestamp = appendInfo.offsetOfMaxTimestamp,records = validRecords)def append(largestOffset: Long,largestTimestamp: Long,shallowOffsetOfMaxTimestamp: Long,records: MemoryRecords): Unit = {if (records.sizeInBytes > 0) {// 添加日志val appendedBytes = log.append(records)}// 当累加超过多少时，才会进行索引的写入// indexIntervalBytes 默认 1048576 字节（1MB）if (bytesSinceLastIndexEntry > indexIntervalBytes) {// 添加偏移量索引offsetIndex.append(largestOffset, physicalPosition)// 添加时间戳索引timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestamp)// 归0bytesSinceLastIndexEntry = 0}// 累加bytesSinceLastIndexEntry += records.sizeInBytes
}// lastOffset + 1
updateLogEndOffset(appendInfo.lastOffset + 1)
2.2.3.1 稀疏索引

kafka 中的偏移量索引和时间戳索引都属于稀疏索引

何为稀疏索引？

正常来说，我们会为每一个日志都创建一个索引，比如：

日志  索引
1     1
2     2
3     3
4     4

但这种方式比较浪费，于是采用稀疏索引，如下：

日志  索引
1     1
2			
3			
4			
5     2
6
7
8

当我们根据偏移量索引查询 1 时，可以查询到日志为 1 的，然后往下遍历搜索想要的即可。

2.2.3.2 偏移量索引
offsetIndex.append(largestOffset, physicalPosition)def append(offset: Long, position: Int) {inLock(lock) {// 索引位置mmap.putInt(relativeOffset(offset))// 日志位置mmap.putInt(position)_entries += 1_lastOffset = offset}
}// 用当前offset减去基准offset
def relativeOffset(offset: Long): Int = {val relativeOffset = offset - baseOffset
}
2.2.3.3 时间戳索引
timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestamp)def maybeAppend(timestamp: Long, offset: Long, skipFullCheck: Boolean = false) {inLock(lock) {if (timestamp > lastEntry.timestamp) {// 添加时间戳mmap.putLong(timestamp)// 添加相对位移(偏移量索引)mmap.putInt(relativeOffset(offset))_entries += 1_lastEntry = TimestampOffset(timestamp, offset)}}
}
2.2.3.4 索引总结

我们的偏移量索引如图下所示：

  • 当我们查询一个消息时，比如消息位移为 23 的
    • 根据二分查找找到偏移量索引下标 22
    • 利用上述我们偏移量 Map 的存储，得到其日志位置 RecordBatch:firstOffset=23 position=762
    • 再根据日志位置，找到真正存储日志的地方

image.png

我们的时间戳索引如图下所示：

  • 基本和我们的偏移量索引类似，只是增加了一层二分查找

image.png

2.2.4 flush刷新

在我们前面添加完之后，我们的数据仅仅是写到 PageCache 里面，需要进行 flush 将其刷新到磁盘中

// 未刷新消息数（unflushedMessages）超过配置的刷新间隔（flushInterval）
if (unflushedMessages >= config.flushInterval){flush()
}def flush() {LogFlushStats.logFlushTimer.time {// 日志刷新log.flush()// 偏移量索引刷新offsetIndex.flush()// 时间戳索引刷新timeIndex.flush()// 事务索引刷新txnIndex.flush()}}

2.3 Follow 获取日志

同样，我们的 Follow 在获取日志时，和我们 Leader 添加日志时一样的方法

image.png

三、流程图

image.png

四、总结

这一篇我们介绍了 Kafka 中日志时如何持久化的以及 Kafka 日志中包括什么数据

鲁迅先生曾说：独行难，众行易，和志同道合的人一起进步。彼此毫无保留的分享经验，才是对抗互联网寒冬的最佳选择。

其实很多时候，并不是我们不够努力，很可能就是自己努力的方向不对，如果有一个人能稍微指点你一下，你真的可能会少走几年弯路。

如果你也对 后端架构 和 中间件源码 有兴趣，欢迎添加博主微信：hls1793929520，一起学习，一起成长

我是爱敲代码的小黄，阿里巴巴淘天集团Java开发工程师，双非二本，培训班出身

通过两年努力，成功拿下阿里、百度、美团、滴滴等大厂，想通过自己的事迹告诉大家，努力是会有收获的！

双非本两年经验，我是如何拿下阿里、百度、美团、滴滴、快手、拼多多等大厂offer的？

我们下期再见。

从清晨走过，也拥抱夜晚的星辰，人生没有捷径，你我皆平凡，你好，陌生人，一起共勉。

这篇关于阿里 P7 三面凉凉，kafka Borker 日志持久化没答上来的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/508821。

相关文章

Spring Boot集成/输出/日志级别控制/持久化开发实践

Spring Boot集成/输出/日志级别控制/持久化开发实践

《SpringBoot集成/输出/日志级别控制/持久化开发实践》SpringBoot默认集成Logback,支持灵活日志级别配置（INFO/DEBUG等）,输出包含时间戳、级别、类名等信息,并可通过... 目录一、日志概述1.1、Spring Boot日志简介1.2、日志框架与默认配置1.3、日志的核心作用
阅读更多...
深度解析Nginx日志分析与499状态码问题解决

深度解析Nginx日志分析与499状态码问题解决

《深度解析Nginx日志分析与499状态码问题解决》在Web服务器运维和性能优化过程中,Nginx日志是排查问题的重要依据,本文将围绕Nginx日志分析、499状态码的成因、排查方法及解决方案展开讨论... 目录前言1. Nginx日志基础1.1 Nginx日志存放位置1.2 Nginx日志格式2. 499
阅读更多...
使用Python构建一个高效的日志处理系统

使用Python构建一个高效的日志处理系统

《使用Python构建一个高效的日志处理系统》这篇文章主要为大家详细讲解了如何使用Python开发一个专业的日志分析工具,能够自动化处理、分析和可视化各类日志文件,大幅提升运维效率,需要的可以了解下... 目录环境准备工具功能概述完整代码实现代码深度解析1. 类设计与初始化2. 日志解析核心逻辑3. 文件处
阅读更多...
C++ Log4cpp跨平台日志库的使用小结

C++ Log4cpp跨平台日志库的使用小结

《C++Log4cpp跨平台日志库的使用小结》Log4cpp是c++类库,本文详细介绍了C++日志库log4cpp的使用方法,及设置日志输出格式和优先级,具有一定的参考价值,感兴趣的可以了解一下... 目录一、介绍1. log4cpp的日志方式2.设置日志输出的格式3. 设置日志的输出优先级二、Window
阅读更多...
怎样通过分析GC日志来定位Java进程的内存问题

怎样通过分析GC日志来定位Java进程的内存问题

《怎样通过分析GC日志来定位Java进程的内存问题》：本文主要介绍怎样通过分析GC日志来定位Java进程的内存问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、GC 日志基础配置1. 启用详细 GC 日志2. 不同收集器的日志格式二、关键指标与分析维度1.
阅读更多...
Python实现对阿里云OSS对象存储的操作详解

Python实现对阿里云OSS对象存储的操作详解

《Python实现对阿里云OSS对象存储的操作详解》这篇文章主要为大家详细介绍了Python实现对阿里云OSS对象存储的操作相关知识,包括连接,上传,下载,列举等功能,感兴趣的小伙伴可以了解下... 目录一、直接使用代码二、详细使用1. 环境准备2. 初始化配置3. bucket配置创建4. 文件上传到os
阅读更多...
解读GC日志中的各项指标用法

解读GC日志中的各项指标用法

《解读GC日志中的各项指标用法》：本文主要介绍GC日志中的各项指标用法,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、基础 GC 日志格式（以 G1 为例）1. Minor GC 日志2. Full GC 日志二、关键指标解析1. GC 类型与触发原因2. 堆
阅读更多...
Redis的持久化之RDB和AOF机制详解

Redis的持久化之RDB和AOF机制详解

《Redis的持久化之RDB和AOF机制详解》：本文主要介绍Redis的持久化之RDB和AOF机制,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录概述RDB（Redis Database）核心原理触发方式手动触发自动触发AOF（Append-Only File）核
阅读更多...
MySQL 打开binlog日志的方法及注意事项

MySQL 打开binlog日志的方法及注意事项

《MySQL打开binlog日志的方法及注意事项》本文给大家介绍MySQL打开binlog日志的方法及注意事项,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要... 目录一、默认状态二、如何检查 binlog 状态三、如何开启 binlog3.1 临时开启（重启后失效）
阅读更多...
SpringBoot项目配置logback-spring.xml屏蔽特定路径的日志

SpringBoot项目配置logback-spring.xml屏蔽特定路径的日志

《SpringBoot项目配置logback-spring.xml屏蔽特定路径的日志》在SpringBoot项目中,使用logback-spring.xml配置屏蔽特定路径的日志有两种常用方式,文中的... 目录方案一：基础配置（直接关闭目标路径日志）方案二：结合 Spring Profile 按环境屏蔽关
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2