KAFKA的ISR的伸缩过程是什么

本篇内容介绍了“KAFKA的ISR的伸缩过程是什么”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这些情况吧!希望大家仔细阅读,能够学有所成!

创新互联是专业的牙克石网站建设公司,牙克石接单;提供成都网站设计、做网站,网页设计,网站设计,建网站,PHP网站建设等专业做网站服务;采用PHP框架,可快速的进行牙克石网站开发网页制作和功能扩展;专业做搜索引擎喜爱的网站,专业的做网站团队,希望更多企业前来合作!

一、前言

知识点总结


二、概述

我们了解ISR列表是不断伸缩的,在副本失效后及时踢出ISR列表,在副本赶上进度之后重新将副本加入到ISR列表中,后面我们就会按照这个思路来看下其中细节。


三、什么是失效副本?

功能失效:节点宕机,在该节点上的副本都属于功能失效副本。
同步失效:follower副本所在的broker因为带宽或者负载等因素无法及时完成同步,导致被踢出ISR。


四、ISR伸缩控制参数了解吗?

在0.9x版本之前,有一个控制参数:replica.lag.max.messages 默认值为4000,表示如果follower的消息个数落后leader个数4000,那么就会被踢出ISR列表;
我们可以想一下这种直接指定条数的方式是否合理呢?显然是不合理的,原因入下:
高吞吐的场景:瞬间就几万条消息,可能follower就滞后个几秒钟就被判定为失效从而被踢出,可能导致ISR列表频繁的变动,以及元数据的频繁更新。
低吞吐的场景:可能一天就几条消息,那可能follower都滞后好几天了依旧存在于ISR中,那ISR不就失去意义了吗?

因此0.9x版本开始,移除了该参数,取而代之的参数是replica.lag.time.max.ms 该参数默认值是10000ms,也就是10s。
也就是说如果follower在10s都没能追上leader的LEO,就会被认定为失效,从而踢出IS列表。


五、ISR是如何将失效副本剔除的?

我们知道了ISR是如何判定失效副本后,再来看下,到底是怎么把这个失效的副本踢出去的呢?
1、每个broker在启动的时候都会启动两个定时任务:

  • isr-expiration:定时检查当前broker上的eader对应的副本失效信息,也就是看当前Leader的ISR列表中是否存在失效副本,默认执行周期为replica.lag.time.max.ms / 2 = 5s

  • isr-change-propagation:定时检查内存isrChangeSet中是否有新的变更数据,固定执行周期为2.5s

2、判断副本失效:
isr-expiration任务会根据当前时间now,减去某follower的 lastCaughtUpTimeMs,如果大于replica.lag.time.max.ms值,则说明失效。
lastCaughtUpTimeMs这个值,在follower的LEO与leader的LEO相等时(Leader中维护了follower的LEO信息),被更新。
也就是说,只有当follower完全追上了Leader才更新,而不是每Fetch一次就更新。

关于为什么不是每次Fetch的时候就更新该值呢?
我们试想一下,如果leader的写入速率远大于follower的同步速率,可能leader已经写了10w条数据了,follower由于网络/负载为原因还在慢悠悠的同步,但是因为Fetch请求是正常发送的,就每次都更新lastCaughtUpTimeMs值,从而认为该follower是有效的,那这不就导致leader和follower之间在这种场景下存在巨大的数据差异了嘛?从而影响数据的可靠性。

3、这个ISR变化的信息如何传递呢?

  1. 由leader所在的broker的isr-expiration定时任务,去检查失效副本和更新zk的/state节点数据,同时写入isrChangeSet

  2. isr-change-propagation去检查isrChangeSet是否有新增数据,如果有,则往zk中的/isr_change_notification节点下创建子节点。

  3. 而Controller对这个节点有一个Watcher,如果发现新增了子节点,那么Controller就会重新从zk中获取到最新的元数据,然后通知所有Broker更新元数据。

从上述过程中,我们还可以知道,实际上这个变更的数据会在内存中停留一段时间,假如这个时候我们对应的broker宕机了,那么不就是改了zk却没有让其他broker更新元数据吗?
其实不是,因为这种情况下,broker宕机会触发controller在zk下的brokers/ids下对应的节点被删除,因此Controller也会让其他的broker更新元数据,所以无论如何都会更新。

最后我们来总结一下整个ISR剔除的过程
每个leader在启动的时候都会启动两个定时检查任务,每隔一段时间检查是否存在失效副本。
如果某个follower的lastCaughtUpTimeMs > 10s那么就会被判定为失效副本
如果定时任务扫描到存在失效副本时,就会往zk的/state节点下更新最新的ISR列表数据,同时将变更数据写入到内存中的isrChangeSet中。
然后另外一个传播任务会定时检查isrChangeSet是否存在需要变更的任务,如果感知到就往zk的/isr_change_notification节点下创建子节点。
最终由Controller感知到节点的变化,然后从zk中获取最新的元数据,然后通知所有的Broker更新元数据,完成整个ISR列表的数据更新。


六、追赶上的副本是如何重新加入ISR中的?

在看完第五小节之后,第六小节就会显得非常简单,无非是需要知道什么时候一个副本会重新判定为同步副本呢? 那就是:当前失效follower的LEO等于leaderHW的时候,即被判断可以重新加入ISR。

那么随之而来的一个问题就是在哪儿去判断followerLEO == leaderHW的呢?
这里和上面的剔除ISR成员不一样,并不是由定时任务去检测的,而是在处理完Fetch请求的时候,如果判断Fetch请求是follower发过来的的(replicaId >= 0),那么就会去看下当前这个follower的LEO是多少(其实就是Fetch请求带过来的),是不是赶上了当前的leaderHW,如果是的那么就执行扩张ISR操作。
扩张ISR操作流程就和上面流程一样了,先写zk下的/state数据,然后写isrChangeSet,最后由Controller感知到数据变化,更新集群元数据。
我们所需要记住的主要差别点在于,ISR列表的扩张是在Fetch请求的时候去判断和执行的。


七、整个伸缩过程总结

最后,我们用图示来加深一点印象。
1、失效副本(图源:《深入理解kafka》)
KAFKA的ISR的伸缩过程是什么

2、踢出ISR列表:
KAFKA的ISR的伸缩过程是什么


八、性能优化

我们由上可知,ISR的伸缩是需要涉及到zk和Controller以及各个Broker的元数据更新的,因此如果太过频繁会造成性能问题。
所以kafka在在判断ISR伸缩之前,还会判断两个条件,以此来降低频率:

  • 上次ISR集合发生变化距离现在已经超过5s。

  • 上一次写入zk的时候,距离现在已经超过60s。

如果一个副本刚追上Leader加入到ISR,但是因为短时间内没有追上LEO,5s之后又被检查到是失效副本,不是又要被踢出去,要更新元数据,这样就太频繁了。 因此就有了上面两个限制,就起码给了多60s的让新加入的follower去追上Leader的LEO。

“KAFKA的ISR的伸缩过程是什么”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站,小编将为大家输出更多高质量的实用文章!


分享文章:KAFKA的ISR的伸缩过程是什么
文章网址:http://pwwzsj.com/article/geeshe.html