[[282170]]

今天先來說說kafka吧，我看Hbase沒什么人看，于是直接跳過，講大家最喜歡的。

一、Kafka架構(gòu)圖

 

 

在一套kafka架構(gòu)中有多個(gè)Producer，多個(gè)Broker,多個(gè)Consumer，每個(gè)Producer可以對(duì)應(yīng)多個(gè)Topic，每個(gè)Consumer只能對(duì)應(yīng)一個(gè)ConsumerGroup。

整個(gè)Kafka架構(gòu)對(duì)應(yīng)一個(gè)ZK集群，通過ZK管理集群配置，選舉Leader，以及在consumer group發(fā)生變化時(shí)進(jìn)行rebalance。

對(duì)于一個(gè)復(fù)雜的分布式系統(tǒng)，如果沒有豐富的經(jīng)驗(yàn)和牛逼的架構(gòu)能力，很難把系統(tǒng)做得簡單易維護(hù)，我們都知道，一個(gè)軟件的生命周期中，后期維護(hù)占了70%，所以系統(tǒng)的可維護(hù)性是極其重要的， kafka 能成為大數(shù)據(jù)領(lǐng)域的事實(shí)標(biāo)準(zhǔn)，很大原因是因?yàn)檫\(yùn)維起來很方便簡單，今天我們來看下 kafka 是怎么來簡化運(yùn)維操作的。

kafka 使用多副本來保證消息不丟失，多副本就涉及到kafka的復(fù)制機(jī)制，在一個(gè)超大規(guī)模的集群中，時(shí)不時(shí)地這個(gè)點(diǎn)磁盤壞了，那個(gè)點(diǎn)cpu負(fù)載高了，出現(xiàn)各種各樣的問題，多個(gè)副本之間的復(fù)制，如果想完全自動(dòng)化容錯(cuò)，就要做一些考量和取舍了。我們舉個(gè)例子說明下運(yùn)維中面對(duì)的復(fù)雜性，我們都知道 kafka 有個(gè) ISR集合，我先說明下這個(gè)概念：

kafka不是完全同步，也不是完全異步，是一種ISR機(jī)制：

1. leader會(huì)維護(hù)一個(gè)與其基本保持同步的Replica列表，該列表稱為ISR(in-sync Replica)，每個(gè)Partition都會(huì)有一個(gè)ISR，而且是由leader動(dòng)態(tài)維護(hù)

2. 如果一個(gè)follower比一個(gè)leader落后太多，或者超過一定時(shí)間未發(fā)起數(shù)據(jù)復(fù)制請求，則leader將其重ISR中移除

3. 當(dāng)ISR中所有Replica都向Leader發(fā)送ACK時(shí)，leader才commit，這時(shí)候producer才能認(rèn)為一個(gè)請求中的消息都commit了。

在這種機(jī)制下，如果一個(gè) producer 一個(gè)請求發(fā)送的消息條數(shù)太多，導(dǎo)致flower瞬間落后leader太多怎么辦?如果 follower不停的移入移出 ISR 會(huì)不會(huì)影響性能?如果對(duì)這種情況加了報(bào)警，就有可能造成告警轟炸，如果我們不加報(bào)警，如果是broker 掛掉或者 broker 因?yàn)镮O性能或者GC問題夯住的情況導(dǎo)致落后leader太多，這種真正需要報(bào)警情況怎么辦呢?

今天我們來看下 kafka 是怎么在設(shè)計(jì)上讓我們完全避免這種運(yùn)維中頭疼的問題的。

二、kafka的復(fù)制機(jī)制

kafka 每個(gè)分區(qū)都是由順序追加的不可變的消息序列組成，每條消息都一個(gè)唯一的offset 來標(biāo)記位置。

 

 

kafka中的副本機(jī)制是以分區(qū)粒度進(jìn)行復(fù)制的，我們在kafka中創(chuàng)建 topic的時(shí)候，都可以設(shè)置一個(gè)復(fù)制因子，這個(gè)復(fù)制因子決定著分區(qū)副本的個(gè)數(shù)，如果leader 掛掉了，kafka 會(huì)把分區(qū)主節(jié)點(diǎn)failover到其他副本節(jié)點(diǎn)，這樣就能保證這個(gè)分區(qū)的消息是可用的。leader節(jié)點(diǎn)負(fù)責(zé)接收producer 打過來的消息，其他副本節(jié)點(diǎn)(follower)從主節(jié)點(diǎn)上拷貝消息。

 

 

kakfa 日志復(fù)制算法提供的保證是當(dāng)一條消息在 producer 端認(rèn)為已經(jīng) committed的之后，如果leader 節(jié)點(diǎn)掛掉了，其他節(jié)點(diǎn)被選舉成為了 leader 節(jié)點(diǎn)后，這條消息同樣是可以被消費(fèi)到的。

這樣的話，leader 選舉的時(shí)候，只能從 ISR集合中選舉，集合中的每個(gè)點(diǎn)都必須是和leader消息同步的，也就是沒有延遲，分區(qū)的leader 維護(hù)ISR 集合列表，如果某個(gè)點(diǎn)落后太多，就從 ISR集合中踢出去。

producer 發(fā)送一條消息到leader節(jié)點(diǎn)后， 只有當(dāng)ISR中所有Replica都向Leader發(fā)送ACK確認(rèn)這條消息時(shí)，leader才commit，這時(shí)候producer才能認(rèn)為這條消息commit了，正是因?yàn)槿绱?，kafka客戶端的寫性能取決于ISR集合中的最慢的一個(gè)broker的接收消息的性能，如果一個(gè)點(diǎn)性能太差，就必須盡快的識(shí)別出來，然后從ISR集合中踢出去，以免造成性能問題。

三、一個(gè)副本怎么才算是跟得上leader的副本

一個(gè)副本不能 “caught up” leader 節(jié)點(diǎn)，就有可能被從 ISR集合中踢出去，我們舉個(gè)例子來說明，什么才是真正的 “caught up” —— 跟leader節(jié)點(diǎn)消息同步。

kafka 中的一個(gè)單分區(qū)的 topic — foo，復(fù)制因子為 3 ，分區(qū)分布和 leader 和 follower 如下圖，現(xiàn)在broker 2和3 是 follower 而且都在 ISR 集合中。我們設(shè)置replica.lag.max.messages 為4，只要 follower 只要不落后leader 大于3條消息，就然后是跟得上leader的節(jié)點(diǎn)，就不會(huì)被踢出去， 設(shè)置replica.lag.time.max.ms 為 500ms， 意味著只要 follower 在每 500ms內(nèi)發(fā)送fetch請求，就不會(huì)被認(rèn)為已經(jīng)dead ，不會(huì)從ISR集合中踢出去。

 

 

現(xiàn)在 producer 發(fā)送一條消息，offset 為3， 這時(shí)候 broker 3 發(fā)生了 GC， 入下圖：

 

 

因?yàn)?broker 3 現(xiàn)在在 ISR 集合中， 所以要么 broker 3 拉取同步上這條 offset 為3 的消息，要么 3 被從 ISR集合中踢出去，不然這條消息就不會(huì) committed， 因?yàn)閞eplica.lag.max.messages=4 為4， broker 3 只落后一條消息，不會(huì)從ISR集合中踢出去， broker 3 如果這時(shí)候 GC 100ms， GC 結(jié)束，然后拉取到 offset 為3的消息，就再次跟 leader 保持完全同步，整個(gè)過程一直在 ISR集合中，如下圖：

 

 

四、什么時(shí)候一個(gè)副本才會(huì)從ISR集合中踢出去

一個(gè)副本被踢出 ISR集合的幾種原因：

• 一個(gè)副本在一段時(shí)間內(nèi)都沒有跟得上 leader 節(jié)點(diǎn)，也就是跟leader節(jié)點(diǎn)的差距大于replica.lag.max.messages， 通常情況是 IO性能跟不上，或者CPU 負(fù)載太高，導(dǎo)致 broker 在磁盤上追加消息的速度低于接收leader 消息的速度。
• 一個(gè) broker 在很長時(shí)間內(nèi)(大于replica.lag.time.max.ms )都沒有向 leader 發(fā)送fetch 請求， 可能是因?yàn)? broker 發(fā)生了 full GC， 或者因?yàn)閯e的原因掛掉了。
• 一個(gè)新 的 broker 節(jié)點(diǎn)，比如同一個(gè) broker id， 磁盤壞掉，新?lián)Q了一臺(tái)機(jī)器，或者一個(gè)分區(qū) reassign 到一個(gè)新的broker 節(jié)點(diǎn)上，都會(huì)從分區(qū)leader 上現(xiàn)存的最老的消息開始同步。

所以說 kafka 0.8 版本后設(shè)置了兩個(gè)參數(shù) ，replica.lag.max.messages 用來識(shí)別性能一直很慢的節(jié)點(diǎn)，replica.lag.time.max.ms 用來識(shí)別卡住的節(jié)點(diǎn)。

五、一個(gè)節(jié)點(diǎn)在什么情況下真正處于落后狀態(tài)

從上面的情況來看，兩個(gè)參數(shù)看似已經(jīng)足夠了，如果一個(gè)副本超過 replica.lag.time.max.ms 還沒有發(fā)送fetch同步請求， 可以認(rèn)為這個(gè)副本節(jié)點(diǎn)卡住了，然后踢出去，但是還有一種比較特殊的情況沒有考慮到。

我們上文中設(shè)置replica.lag.max.messages 為4，之所以設(shè)置為 4， 是我們已經(jīng)知道 producer 每次請求打過來的消息數(shù)都在 4 以下，如果我們的參數(shù)是作用于多個(gè) topic 的情況，那么這個(gè) producer 最大打過來的消息數(shù)目就不好估計(jì)了，或者說在經(jīng)常出現(xiàn)流量抖動(dòng)的情況下，就會(huì)出現(xiàn)一個(gè)什么情況呢，我們還是使用例子說明：

如果我們的 topic — foo 的 producer 因?yàn)榱髁慷秳?dòng)打過來一個(gè) 包含 4條消息的請求，我們設(shè)置的replica.lag.max.messages 還是為4， 這個(gè)時(shí)候，所有的 follower 都會(huì)因?yàn)槌雎浜髼l數(shù)被踢出 ISR集合：

 

 

然后，因?yàn)?follower 是正常的，所以下一次 fetch 請求就會(huì)又追上 leader， 這時(shí)候就會(huì)再次加入 ISR 集合，如果經(jīng)常性的抖動(dòng)，就會(huì)不斷的移入移出ISR集合，會(huì)造成令人頭疼的 告警轟炸。

 

 

這里的核心問題是，在海量的 topic 情況下，或者經(jīng)常性的流量抖動(dòng)情況下，我們不能對(duì) topic 的producer 每次打過來的消息數(shù)目做任何假設(shè)，所以就不太好定出來一個(gè) 合適的eplica.lag.max.messages值

六、一個(gè)配置全部搞定

其實(shí)只有兩種情況是異常的，一種就是卡住，另外一種是follower 性能慢，如果我們只根據(jù) follower 落后 leader 多少來判斷是否應(yīng)該把 follower 提出ISR集合，就必須要對(duì)流量進(jìn)行預(yù)測估計(jì)，怎么才能避免這種不靠譜的估計(jì)呢?

kafka 給出的方案是這樣的：

對(duì) replica.lag.time.max.ms 這個(gè)配置的含義做了增強(qiáng)，和之前一樣，如果 follower 卡住超過這個(gè)時(shí)間不發(fā)送fetch請求， 會(huì)被踢出ISR集合，新的增強(qiáng)邏輯是，在 follower 落后 leader 超過eplica.lag.max.messages 條消息的時(shí)候，不會(huì)立馬踢出ISR 集合，而是持續(xù)落后超過replica.lag.time.max.ms 時(shí)間，才會(huì)被踢出，這樣就能避免流量抖動(dòng)造成的運(yùn)維問題，因?yàn)閒ollower 在下一次fetch的時(shí)候就會(huì)跟上leader， 這樣就也不用對(duì) topic 的寫入速度做任何的估計(jì)嘍。