山东住房与城乡建设部网站,重庆承越网站建设公,自己写的html放入wordpress,哈尔滨建站在线咨询杂 在0.9.0.0之前#xff0c;Kafka提供了replica lag.max.messages 来控制follower副本最多落后leader副本的消息数量#xff0c;follower 相对于leader 落后当超过这个数量的时候就判定该follower是失效的#xff0c;就会踢出ISR#xff0c;这里的指的是具体的LEO值。 对…杂 在0.9.0.0之前Kafka提供了replica lag.max.messages 来控制follower副本最多落后leader副本的消息数量follower 相对于leader 落后当超过这个数量的时候就判定该follower是失效的就会踢出ISR这里的指的是具体的LEO值。 对应的Kafka 也针对这些场景提供了一些控制的参数前面提到的replica.lag.max.message以数量为标准衡量是否落后还有以时间为衡量标准的replica.lag.time.max多久没有向leader 请求数据 这些是0.9.0.0之前的版本这个实现是可以适应大多数环境的但是存在一个严重的缺陷当qps持续上升请求打满之后很容易造成同步速率下降或者长时间无响应进而导致很多follower被踢出ISR在流量高峰时期会挺常见这就导致使用者需要在不同的场景定制不同的参数配置但是什么时候有突发流量什么时候去配置并且令其生效这个事儿不现实所以说Kafka这一点算是一个缺陷吧。 0.9.0.0 之后提供了一个更加适合的方式来解决这个问题采用Kafka 落后于消费进度的时间长度来判断是否踢出ISR这样有效的避免了在突发流量偶然落后于leader 被不合理的踢出ISR的情况如果长时间落后于leader 这种情况实际故障是需要去踢的也没问题也就有效的避免了ISR的反复移进移出所带来的代价。 Replica leader分区会维护自身本地副本以及所有follower副本远程副本的相关状态而follower分区只维护自己的状态本地副本。 本地副本的LEO和HW都会更新远程副本的LEO会更新HW不会被更新。Leader分区之所以要维护远程副本是为了帮助确定HW。LEO和HW的更新时机 更新对象更新时机leader分区本地副本LEO接收到生产者发送的消息写入本地磁盘后会更新LEOleader分区远程副本LEOfollower从leader拉取消息时会告诉leader从哪个位移开始拉这个位置就会更新到远程副本的LEOfollower分区本地副本LEO从leader分区拉取消息写入本地磁盘后会更新LEOleader分区本地副本HW1. 更新本地副本LEO后2. 更新远程副本LEO后。取本地副本和远程副本LEO中的最小值leader分区远程副本HW不会更新从leader分区拉取消息写入本地磁盘后会更新LEO比较LEO和leader发来的HW取两者最小值更新为HW 字段 brokerIdbrokerIdtopicPartition类型为TopicPartition副本对应的分区log副本对应的Log对象远程副本的此字段为空通过此字段区分是本地副本还是远程副本highWatermarkMetadata记录HW的值logEndOffsetMetadata本地副本对应LEO值logs end offset远程副本该值只在follower fetch的时候更新logStartOffset本地副本对应LSOlogs start offset远程副本该值只在follower fetch的时候更新lastFetchLeaderLogEndOffsetleader收到follower的FetchRequest时候的LEO值用来确定follower的lastCaughtUpTimeMslastFetchTimeMsleader收到follower的FetchRequest时候的时间用来确定follower的lastCaughtUpTimeMslastCaughtUpTimeMs该follower的LEO大于等于此时刻leader的LEO用来确定该follower相对于该分区ISR的lag方法 // 通过有无log判断是本地副本还是远程副本 def isLocal: Boolean log.isDefined // 获取lastCaughtUpTimeMs def lastCaughtUpTimeMs _lastCaughtUpTimeMs // def updateLogReadResult(logReadResult: LogReadResult) {...}// 对于本地副本不能直接更新LEO其LEO由Log.logEndOffsetMetadata字段决定private def logEndOffset_(newLogEndOffset: LogOffsetMetadata) {if (isLocal) {throw new KafkaException(sxxx)} else {logEndOffsetMetadata newLogEndOffsettrace(sxxx)}}// 本地副本和远程副本的LEO获取方式也不同def logEndOffset: LogOffsetMetadata if (isLocal)log.get.logEndOffsetMetadataelselogEndOffsetMetadata // LSO的set和get方法与LEO相同此处省略 // 只有本地副本可以更新HWdef highWatermark_(newHighWatermark: LogOffsetMetadata) {if (isLocal) {if (newHighWatermark.messageOffset 0)throw new IllegalArgumentException(High watermark offset should be non-negative)highWatermarkMetadata newHighWatermarklog.foreach(_.onHighWatermarkIncremented(newHighWatermark.messageOffset))} else {throw new KafkaException(sShould not set high watermark on partition $topicPartitions non-local replica $brokerId)}}Partition Partition负责Replica对象的管理和维护包括副本角色切换、ISR集合管理等。 字段 topic和partitionId此Partition对象代表的Topic名称和分区编号。localBrokerId当前Broker的id可以与replicaId比较从而判断指定的Replica是否表示本地副 本。logManager当前Broker上的LogManager对象。 zkClient操作ZooKeeper的辅助类。leaderEpoch该分区Leader副本的年代信息。leaderReplicaIdOpt该分区的Leader副本所在broker的id。inSyncReplicasSet[Replica]类型该集合维护了该分区的ISR集合ISR集合是AR集合的子集。allReplicasMapPool[Int, Replica]类型维护了该分区的全部副本的集合AR集合的信 息。Partition中的方法按照功能可以划分为下列5类 获取或创建ReplicagetOrCreateReplica()方法副本的Leader/Follower角色切换makeLeader()方法和makeFollower()方法ISR集合管理maybeExpandIsr()方法和maybeShrinkIsr()方法调用日志存储子系统完成消息写入appendRecordsToLeader()方法检测HW的位置checkEnoughReplicasReachOffset()方法 上述五类方法为ReplicaManager的实现提供了基础支持。其他较为简单的辅助方法不再做详细介绍请 读者参考源码学习。 获取或创建Replicadone getOrCreateReplica()方法主要负责在AR集合assignedReplicaMap 中查找指定副本的Replica对象 如 果查找不到则创建Replica对象并添加到AR集合中管理。 如果创建的是Local Replica 还会创建或恢复 对应的Log并初始化或恢复 HW。 HW与Log. recoveryPoint类似 也会需要记录到文件中保存 在每个log 目录下都有一个replication-offset-checkpoint文件记录了此目录下每个分区的HW。 在ReplicaManager启动时 会读取此文件到highWatermarkCheckpoints这个Map中 之后会定时更新replication-offset-checkpoint文件。 副本角色切换 Broker会根据KafkaController发送的LeaderAndISRRequest请求控制副本的Leader/Follower角色切换。 Partition.makeLeader()方法是处理LeaderAndISRRequest中比较重要的环节之一 它会将Local Replica设置成 Leader副本。Partition.makeFollower()方法与Partition.makeLeader()方法类似 也是处理LeaderAndISRRequest的环节之一。 它的功能是按照PartitionState指定的信息 将Local Replica设置为Follower副本。 ISR集合管理done Partition除了对副本的Leader/Follower角色进行管理 还需要管理ISR集合。 随着Follower副本不断与Leader副本进行消息同步 Follower副本的LEO会逐渐后移 并最终追赶上Leader副本的LEO 此时该Follower副本就有资格进入ISR集合。 Partition.maybeExpandIsr()方法实现了扩张ISR集合的功能KafkaApis.handleFetchRequest()处理fetch请求的时候会判断该fetch是否来自follower如果来自follower则会调用Partition.updateFollowerLogReadResults() - Partition.maybeExpandIsr()。 在ReplicaManager中使用定时任务周期性地调用maybeShrinkIsr ()方法检查ISR集合中Follower副本与Leader副本之间的同步差距 并对ISR集合进行缩减。 有一点需要读者注意 在ISR集合发生增减的时候 都会将最新的ISR集合保存在ZooKeeper中 具体的保存路是/brokers/topics/[topic_name]/partitions/[partitionId]/state。 后面介绍的KafkaController会监听此路径中数据的变化 追加消息done 调用日志存储子系统完成消息写入比较简单后续补充。 内部会调用Log.appendAsLeader()执行真正的写入操作。 然后调用ReplicaManager.tryCompleteDelayedFetch()尝试完成DelayedFetch。 然后调用maybeIncrementLeaderHW()尝试更新高水位HWISR可能缩容为1这时HW就会更新。 如果高水位HW有变动则尝试完成所有的Delay操作DelayedFetch、DelayedProduce、DelayedDeleteRecords。 检测HW的位置done 在检测DelayedProduce的执行条件时 简单提到了Partition.checkEnoughReplicasReachOffset()方法 此方法会检测其参数指定的消息是否已经被ISR集合中所有Follower副本同步。 该方法会判断当前leader副本的HW是否已经大于等于传入的偏移量如果是则说明已经同步返回true和0错误码否则还没有同步返回false和0错误码。注意当某个topic设置了min.insync.replicas参数如果insync个数不满足但是HW已经满足则会返回true和一个20错误码。 ReplicaManager ReplicaManager的功能是管理一个Broker范围内的Partition信息。ReplicaManager的实现依赖于日志存储子系统、DelayedOperationPurgatory、KafkaScheduler等组件底层依赖于Partition和Replica。 字段 logManagerLogManager对象对分区的读写操作都委托给底层的日志存储子系统。schedulerKafkaScheduler对象用于执行ReplicaManager中的周期性定时任务。在ReplicaManager 中总共有4个周期性任务它们分别是highwatermark-checkpoint任务、isr-expiration任务、isrchange- propagation、shutdown-idle-replica-alter-log-dirs-thread任务。controllerEpoch记录KafkaController的年代信息当重新选举Controller Leader时该字段值会递 增。之后在ReplicaManager处理来自KafkaController的请求时会先检测请求中携带的年代信息 是否等于controllerEpoch字段的值这就避免接收旧Controller Leader发送的请求。这种设计方式在 分布式系统中比较常见。localBrokerId当前Broker的id主要用于查找Local Replica。allPartitionsPool[(String, Int), Partition]类型其中保存了当前Broker上分配的所有Partition信息。replicaFetcherManager在ReplicaFetcherManager中管理了多个ReplicaFetcherThread线程 ReplicaFetcherThread线程会向Leader副本发送FetchRequest请求来获取消息实现Follower副本与 Leader副本同步。ReplicaFetcherManager对象在ReplicaManager初始化时被创建后面会详细介绍 ReplicaFetcherManager与ReplicaFetcherThread的功能。highWatermarkCheckpointsMap[String, OffsetCheckpoint]类型用于缓存每个log目录与 OffsetCheckpoint之间的对应关系OffsetCheckpoint记录了对应log目录下的replication-offset-checkpoint文件该文件中记录了data目录下每个Partition的HW。ReplicaManager中的 highwatermark-checkpoint任务会定时更新replication-offset-checkpoint文件的内容。isrChangeSetSet[TopicAndPartition]类型用于记录ISR集合发生变化的分区信息。delayedProducePurgatory、 delayedFetchPurgatory用于管理DelayedProduce和DelayedFetch的 DelayedOperationPurgatory对象。zkClient操作ZooKeeper的辅助类。角色切换 在Kafka集群中会选举一个Broker成为KafkaController的Leader 它负责管理整个Kafka集群。 Controller Leader根据Partition的Leader副本和Follower副本的状态向对应的Broker节点发送LeaderAndIsrRequest 这个 请求主要用于副本的角色切换 即指导Broker将其上的哪些分区的副本切换成Leader角色 哪些分区的副本切换成Follower角色。 LeaderAndIsrRequest首先由KafkaAPis.handleLeaderAndIsrRequest()方法进行处理 其核心逻辑是通过 ReplicaManager提供的becomeLeaderOrFollower()方法实现的 而becomeLeaderOrFollower()又依赖于上一小节介绍的Partition.makeLeader()方法和makeFollower()方法 调用链路 追加/读取消息done 当Local Replica切换为Leader副本之后 就可以处理生产者发送的ProducerRequest 将消息写入到Log中。 调用链路KafkaApis.handleProduceRequest() - ReplicaManager.appendRecords() - ReplicaManager.appendToLocalLog() - Partition.appendRecordsToLeader() - Log.appendAsLeader() 主要逻辑在 Partition.appendRecordsToLeader()中之前已经分析不再展开。 Leader副本的另一个重要功能是处理FetchRequest进行消息读取。 调用链路KafkaApis.handleFetchRequest() - ReplicaManager.fetchMessages() - ReplicaManager.readFromLocalLog() - Log.read() 这里主要分析readFromLocalLog()方法在该方法中会循环遍历拉取所有指定分区中的数据。fetch请求中会指定两个参数一个是单次最多拉取多少数据一个是单次单分区最多拉取多少数据对于follower的fetch这两个默认值分别为10MB和1MB配置项为replica.fetch.response.max.bytes和replica.fetch.max.bytes对于消费者客户端还未确认todo。因此每个分区最多拉取1MB当从多个分区中累计拉取到10MB后就会返回。另外需要注意当要读取的分区中的单条消息大于1MB时如果已经从其他分区读到了数据则不会再读取否则会读取一条大消息。 副本同步done Follower副本与Leader副本同步的功能由ReplicaFetcherManager组件实现。具体的同步逻辑交由ReplicaFetcherThread线程处理。 AbstractFetcherManager是ReplicaFetcherManager的抽象类它的addFetcherForPartitions()方法中会为分区添加fetch线程每个broker的fetch线程个数由num.replica.fetchers确定默认为1。注意这里的fetch线程个数是向单个broker同步数据的线程数实际环境中都是向n个broker拉取数据的则真实fetch线程个数是num.replica.fetchers乘以n。比如3个节点的kafkakafka0会起1个fetch1线程从kafka1中拉取消息起1个fetch2线程从kafka2中拉取消息。 还要注意num.replica.fetchers的值并不是真正的fetch线程个数下面的方法是将某个分区分配给某个fetcher线程的代码。可以看到是根据topic的hash值和partitionId确定一个key然后根据该key查找map中对应的fetcher线程没有则新建进行关联。首先fetcher线程个数最多为分区个数即使我们设置了num.replica.fetchers为10000也不会有10000个fetch线程其次即使num.replica.fetchers远小于分区数实际fetcher线程数可能比num.replica.fetchers更少。试想这样一种场景num.replica.fetchers为12Utils.abs(31 * topic.hashCode() partitionId) % numFetchersPerBroker中的取值没有3和4则只会有10个fetcher线程。 private[server] def getFetcherId(topic: String, partitionId: Int) : Int {lock synchronized {Utils.abs(31 * topic.hashCode() partitionId) % numFetchersPerBroker}}分区和fetch线程对应后就会启动该fetch线程。 核心业务代码在AbstractFetcherThread的doWork()方法中 override def doWork() {maybeTruncate()val fetchRequest inLock(partitionMapLock) {val ResultWithPartitions(fetchRequest, partitionsWithError) buildFetchRequest(states)if (fetchRequest.isEmpty) {trace(sThere are no active partitions. Back off for $fetchBackOffMs ms before sending a fetch request)partitionMapCond.await(fetchBackOffMs, TimeUnit.MILLISECONDS)}handlePartitionsWithErrors(partitionsWithError)fetchRequest}if (!fetchRequest.isEmpty)processFetchRequest(fetchRequest)}主要是两个方法buildFetchRequest()和processFetchRequest()。 buildFetchRequest()是构造拉取请求有两个参数值得注意一个是replica.fetch.response.max.bytes指定了单次最多拉取多少数据默认是10MB一个是replica.fetch.max.bytes指定了单次单分区最多拉取多少数据默认1MB。 processFetchRequest()是发送请求并对响应进行处理主要是两个抽象方法fetch()和processPartitionData()。均在ReplicaFetcherThread中实现。fetch()中通过ReplicaFetcherBlockingSend.sendRequest()实现请求的发送并拿到响应在具体实现中发送完响应后会一直在while循环中执行client.poll()方法等待直到拿到响应。processPartitionData()是将拿到的响应数据追加到本地Log并更新follower副本的HW字段。 在正常逻辑下fetch()会调用processPartitionData()方法追加数据如果在fetch()过程中遇到了一些异常情况leader分区会返回错误码Errors.OFFSET_OUT_OF_RANGEfetch()会调用handleOffsetOutOfRange()方法进行处理。 Errors.OFFSET_OUT_OF_RANGE对应两种情况 一种是follower的LEO小于leader的logStartOffset。出现的场景follower下线很久后上线此时leader的老数据日志已经删了很多当前的logStartOffset大于follower的LEO。A一种是follower的LEO大于leader的LEO。出现的场景follower下线leader继续写入消息follower上线开始同步消息但还没同步到能进入ISR集合此时ISR集合中的副本全部下线follower变成了leader旧leader重新上线后变成follower此时follower的LEO大于新leader的LEO**B** handleOffsetOutOfRange()在实际处理时会重新发送一个请求获取leader分区的LEO在此时间段内leader分区可能不断有消息写入因此第2种情况在当下处理的时候又会变为两种情况 和之前一致follower的LEO大于leader的LEO**B1**因为leader分区不断写入消息此时follower的LEO已经小于leader的LEO**B2** 对于情形B1数据会截断到leader的LEO并重新发送fetch请求offset以leader的LEO为准。对于情形B2会重新发送fetch请求offset以follower的LEO为准。对于情形A会删除所有的数据日志并重新发送fetch请求以leader的logStartOffset为准。 注意对于情形B1和B2都是由于unclean leader election的场景引起的都有可能出现副本中某一段数据不一致的情况。在2.0.1版本中没做处理 副本同步全流程 对于服务端来说如果follower的拉取请求过来时没有数据可以返回则会构造DelayedFetch请求。一方面会放入SystemTimer中超时后会返回。另一方面会放入Watchers中等待触发完成时机。 触发时机主分区中有数据写入时。 对于服务端来说客户端的生产请求过来当ack-1时会生成DelayedProduce需要等待follower同步成功后才能返回响应。DelayedProduce也会放入SystemTimer和Watchers中。 触发时机接收到follower的fetch请求或者分区的HW发生了变化 时序如下 服务端处理客户端发送的生产请求服务端生成DelayedProduce等待follower同步数据follower发送fetch请求请求消息数据服务端接收fetch请求获得follower当前的LEO更新HW判断DelayedProduce当前还未同步成功follower拿到消息数据返回追加到自己的Log中然后继续发送下一个fetch请求服务端接收fetch请求获得follower当前的LEO更新HW判断DelayedProduce已经同步成功完成DelayedProduce放入responseQueue中。 假设某个时刻leader的HW和LEO都为1000follower的LEO也为1000。生产者单次请求写入了2条消息。 关闭副本done 当Broker接收到来自KafkaController的StopReplicaRequest请求时 会关闭其指定的副本 并根据 StopReplicaRequest中的字段决定是否删除副本对应的Log。 在分区的副本进行重新分配、 关闭Broker等过程中都会使用到此请求 但是需要注意的是 StopReplicaRequest并不代表一定会删除副本对应的Log 例如shutdown的场景下就没有必要删除Log。 而在重新分配Partition副本的场景下 就需要将旧副本及其Log删除。 定时任务done highwatermark-checkpoint任务会周期性地记录每个Replica的HW并保存到其log目录中的replicationoffset-checkpoint文件中。 isr-expiration任务会周期性地调用maybeShrinkIsr()方法检测每个分区是否需要缩减其ISR集合。 isr-change-propagation任务会周期性地将ISR集合发生变化的分区记录到ZooKeeper中。 highwatermark-checkpoint 这个定时任务是在ReplicaManager.becomeLeaderOrFollower()中启动的。目的是确保所有的分区都已经完全populated来避免奇怪的race conditions。 运行间隔由配置项replica.high.watermark.checkpoint.interval.ms指定默认为5000ms。 主体逻辑在ReplicaManager.checkpointHighWatermarks()方法中实现。 // Flushes the highwatermark value for all partitions to the highwatermark filedef checkpointHighWatermarks() {val replicas nonOfflinePartitionsIterator.flatMap { partition val replicasList: mutable.Set[Replica] mutable.Set()partition.getReplica(localBrokerId).foreach(replicasList.add)partition.getReplica(Request.FutureLocalReplicaId).foreach(replicasList.add)replicasList}.filter(_.log.isDefined).toBuffer// 获取全部的Replica对象按照副本所在的log目录进行分组val replicasByDir replicas.groupBy(_.log.get.dir.getParent)for ((dir, reps) - replicasByDir) {// 获取当前log目录下的全部副本的HWval hwms reps.map(r r.topicPartition - r.highWatermark.messageOffset).toMaptry {// 将HW更新到log目录下的replication-offset-checkpoint文件中highWatermarkCheckpoints.get(dir).foreach(_.write(hwms))} catch {case e: KafkaStorageException error(sError while writing to highwatermark file in directory $dir, e)}}}isr-change-propagation、isr-expiration和shutdown-idle-replica-alter-log-dirs-thread 这3个定时任务是kafka启动的时候就开始的。具体的调用栈为 KafkaServer.startup() - ReplicaManager.startup()。 isr-change-propagation运行间隔为2500ms。 isr-expiration运行间隔由replica.lag.time.max.ms/2指定默认为10000/2 ms。也即一个follower分区在已经落后之后最多可以在isr中存在1.5倍的replica.lag.time.max.ms时间。内部调用Partition.maybeShrinkIsr()方法。 shutdown-idle-replica-alter-log-dirs-thread运行间隔为10000ms。 MetadataCachedone MetadataCache是Broker用来缓存整个集群中全部分区状态的组件。 KafkaController通过向集群中的Broker发送UpdateMetadataRequest来更新其MetadataCache中缓存的数据 每个Broker在收到该请求后会异步更新MetadataCache中的数据。 字段 cache Map[String,Map[Int, UpdateMetadataRequest.PartitionState]]类型 记录了每个分区的状态 其中使用PartitionState记录Partition的状态。外层map的key为topic内层map的key为分区号。aliveBrokers Map[Int, Broker]类型 记录了当前可用的Broker信息 其中使用Broker类记录每个存活Broker的网络位置信息host、 ip、 port等 。aliveNodes Map[Int,Map[ListenerName, Node]]类型 记录了可用节点的信息UpdateMetadataRequest由KafkaApis.handleUpdateMetadataRequest()方法处理 它直接将请求交给ReplicaManager.maybeUpdateMetadataCache()方法处理。 MetadataCache.updateCache()方法中完成了对aliveBrokers、aliveNodes、 cache字段的更新。 生产者和消费者中使用Metadata对象缓存Kafka集群的元信息 在 Metadata更新时会向服务端发送MetadataRequest。 MetadataRequest首先由KafkaApis. handleTopicMetadataRequest()方法进行处理。 在KafkaApis.getTopicMetadata()方法中完成对MetadataCache的查询 同时还会根据配置以及Topic的名称决定是否自动创建未知MetadataCache查找不到 的Topic。 总结 num.replica.fetchers 单个broker的拉取线程默认1 replica.fetch.response.max.bytes 单次最多拉取多少数据默认10MB replica.fetch.max.bytes 单次单分区最多拉取多少数据默认1MB