滴滴二面之Kafka如何读写副本消息的

滴滴二面之Kafka如何读写副本消息的

目录前言appendRecords-副本写入副本读取：fetchMessages总结

前言

无论是读取副本还是写入副本，都是通过底层的Partition对象完成的，而这些分区对象全部保存在上节课所学的allPartitions字段中。可以说，理解这些字段的用途，是后续我们探索副本管理器类功能的重要前提。

现在，我们就来学习下副本读写功能。整个Kafka的同步机制，本质上就是副本读取+副本写入，搞懂了这两个功能，你就知道了Follower副本是如何同步Leader副本数据的。

appendRecords-副本写入

向副本底层日志写入消息的逻辑就实现在ReplicaManager#appendRecords。

Kafka需副本写入的场景：

生产者向Leader副本写入消息Follower副本拉取消息后写入副本仅该场景调用Partition对象的方法，其余3个都是调用appendRecords完成消费者组写入组信息事务管理器写入事务信息（包括事务标记、事务元数据等）

appendRecords方法将给定的一组分区的消息写入对应Leader副本，并根据PRODUCE请求中acks的设置，有选择地等待其他副本写入完成。然后，调用指定回调逻辑。

appendRecords向副本日志写入消息的过程：

执行流程

可见，appendRecords：

实现消息写入的方法是appendToLocalLog

判断是否需要等待其他副本写入的方法delayedProduceRequestRequired

appendToLocalLog写入副本本地日志

利用Partition#appendRecordsToLeader写入消息集合，就是利用appendAsLeader方法写入本地日志的。

delayedProduceRequestRequired

判断消息集合被写入到日志之后，是否需要等待其它副本也写入成功：

private def delayedProduceRequestRequired(

requiredAcks: Short,

entriesPerPartition: Map[TopicPartition, MemoryRecords],

localProduceResults: Map[Tophttp://icPartition, LogAppendResult]): Boolean = {

requiredAcks == -1 && entriesPerPartition.nonEmpty &&

localProduceResults.values.count(_.exception.isDefined) < entriesPerPartition.size

}

若等待其他副本的写入，须同时满足：

requiredAcks==-1依然有数据尚未写完至少有一个分区的消息，已成功被写入本地日志

2和3可结合来看。若所有分区的数据写入都不成功，则可能出现严重错误，此时应不再等待，而是直接返回错误给发送方。

而有部分分区成功写入，部分分区写入失败，则可能偶发的瞬时错误导致。此时，不妨将本次写入请求放入Purgatory，给个重试机会。

副本读取：fetchMessages

ReplicaManager#fetchMessages负责读取副本数据。无论：

java消费者APIFollower副本

拉取消息的主途径都是向Broker发FETCH请求，Broker端接收到该请求后，调用fetchMessages从底层的Leader副本取出消息。

fetchMessages也可能会延时处理FETCH请求，因Broker端必须要累积足够多数据后，才会返回Response给请求发送方。

整个方法分为：

读取本地日志

首先判断，读取消息的请求方，就能确定可读取的范围了。

fetchIsolation，读取隔离级别:

对Follower副本，它能读取到Leader副本LEO值以下的所有消息普通Consumer，只能“看到”Leader副本高水位值以下的消息

确定可读取范围后，调用readFromLog读取本地日志上的消息数据，并将结果赋给logReadResults变量。readFromLog调用readFromLocalLog，在待读取分区上依次调用其日志对象的read方法执行实际的消息读取。

根据读取结果确定Response

根据上一步读取结果创建对应Response：

根据上一步得到的读取结果，统计可读取的总字节数，然后判断此时是否能够立即返回Reponse。

副本管理器读写副本的两个方法appendRecords和fetchMessages本质上在底层分别调用Log的append和read方法，以实现本地日志的读写操作。完成读写操作后，这两个方法还定义了延时处理的条件。一旦满足延时处理条件，就交给对应Purgatory处理。

从这俩方法可见单个组件融合一起的趋势。虽然我们学习单个源码文件的顺序是自上而下，但串联Kafka主要组件功能的路径却是自下而上。

如副本写入操作，日志对象append方法被上一层的Partition对象中的方法调用，而后者又进一步被副本管理器中的方法调用。我们按自上而下阅读了副本管理器、日志对象等单个组件的代码，了解了各自的独立功能。

现在开始慢慢地把它们融合一起，构建Kafka操作分区副本日志对象的完整调用路径。同时采用这两种方式来阅读源码，就能更高效弄懂Kafka原理。

总结

Kafka副本状态机类ReplicaMDkXOzOanager读写副本的核心方法：

appendRecords：向副本写入消息，利用Log#append方法和Purgatory机制实现Follower副本向Leader副本获取消息后的数据同步操作fetchMessages：从副本读取消息，为普通Consumer和Follower副本所使用。当它们向Broker发送FETCH请求时，Brokhttp://er上的副本管理器调用该方法从本地日志中获取指定消息

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。