資料內(nèi)容：

Kafka核心總控制器Controller

在Kafka集群中會(huì)有一個(gè)或者多個(gè)broker，其中有一個(gè)broker會(huì)被選舉為控制器（Kafka Controller），它負(fù)責(zé)管理整個(gè)
集群中所有分區(qū)和副本的狀態(tài)。
當(dāng)某個(gè)分區(qū)的leader副本出現(xiàn)故障時(shí)，由控制器負(fù)責(zé)為該分區(qū)選舉新的leader副本。
當(dāng)檢測(cè)到某個(gè)分區(qū)的ISR集合發(fā)生變化時(shí)，由控制器負(fù)責(zé)通知所有broker更新其元數(shù)據(jù)信息。
當(dāng)使用kafka-topics.sh腳本為某個(gè)topic增加分區(qū)數(shù)量時(shí)，同樣還是由控制器負(fù)責(zé)讓新分區(qū)被其他節(jié)點(diǎn)感知
到。
 

Controller選舉機(jī)制

在kafka集群?jiǎn)?dòng)的時(shí)候，會(huì)自動(dòng)選舉一臺(tái)broker作為controller來(lái)管理整個(gè)集群，選舉的過(guò)程是集群中每個(gè)broker都會(huì)
嘗試在zookeeper上創(chuàng)建一個(gè) /controller 臨時(shí)節(jié)點(diǎn)，zookeeper會(huì)保證有且僅有一個(gè)broker能創(chuàng)建成功，這個(gè)broker
就會(huì)成為集群的總控器controller。
當(dāng)這個(gè)controller角色的broker宕機(jī)了，此時(shí)zookeeper臨時(shí)節(jié)點(diǎn)會(huì)消失，集群里其他broker會(huì)一直監(jiān)聽(tīng)這個(gè)臨時(shí)節(jié)
點(diǎn)，發(fā)現(xiàn)臨時(shí)節(jié)點(diǎn)消失了，就競(jìng)爭(zhēng)再次創(chuàng)建臨時(shí)節(jié)點(diǎn)，就是我們上面說(shuō)的選舉機(jī)制，zookeeper又會(huì)保證有一個(gè)broker
成為新的controller。
具備控制器身份的broker需要比其他普通的broker多一份職責(zé)，具體細(xì)節(jié)如下：
1. 監(jiān)聽(tīng)broker相關(guān)的變化。為Zookeeper中的/brokers/ids/節(jié)點(diǎn)添加BrokerChangeListener，用來(lái)處理broker
增減的變化。
2. 監(jiān)聽(tīng)topic相關(guān)的變化。為Zookeeper中的/brokers/topics節(jié)點(diǎn)添加TopicChangeListener，用來(lái)處理topic增減
的變化；為Zookeeper中的/admin/delete_topics節(jié)點(diǎn)添加TopicDeletionListener，用來(lái)處理刪除topic的動(dòng)作。
3. 從Zookeeper中讀取獲取當(dāng)前所有與topic、partition以及broker有關(guān)的信息并進(jìn)行相應(yīng)的管理。對(duì)于所有topic
所對(duì)應(yīng)的Zookeeper中的/brokers/topics/[topic]節(jié)點(diǎn)添加PartitionModificationsListener，用來(lái)監(jiān)聽(tīng)topic中的
分區(qū)分配變化。
4. 更新集群的元數(shù)據(jù)信息，同步到其他普通的broker節(jié)點(diǎn)中。
 

Partition副本選舉Leader機(jī)制

controller感知到分區(qū)leader所在的broker掛了(controller監(jiān)聽(tīng)了很多zk節(jié)點(diǎn)可以感知到broker存活)，controller會(huì)從
ISR列表(參數(shù)unclean.leader.election.enable=false的前提下)里挑第一個(gè)broker作為leader(第一個(gè)broker最先放進(jìn)ISR
列表，可能是同步數(shù)據(jù)最多的副本)，如果參數(shù)unclean.leader.election.enable為true，代表在ISR列表里所有副本都掛
了的時(shí)候可以在ISR列表以外的副本中選leader，這種設(shè)置，可以提高可用性，但是選出的新leader有可能數(shù)據(jù)少很多。
副本進(jìn)入ISR列表有兩個(gè)條件：
1. 副本節(jié)點(diǎn)不能產(chǎn)生分區(qū)，必須能與zookeeper保持會(huì)話以及跟leader副本網(wǎng)絡(luò)連通
2. 副本能復(fù)制leader上的所有寫(xiě)操作，并且不能落后太多。(與leader副本同步滯后的副本，是由
replica.lag.time.max.ms 配置決定的，超過(guò)這個(gè)時(shí)間都沒(méi)有跟leader同步過(guò)的一次的副本會(huì)被移出ISR列表)
 

消費(fèi)者消費(fèi)消息的offset記錄機(jī)制

每個(gè)consumer會(huì)定期將自己消費(fèi)分區(qū)的offset提交給kafka內(nèi)部topic：__consumer_offsets，提交過(guò)去的時(shí)候，key是
consumerGroupId+topic+分區(qū)號(hào)，value就是當(dāng)前offset的值，kafka會(huì)定期清理topic里的消息，最后就保留最新的
那條數(shù)據(jù)
因?yàn)開(kāi)_consumer_offsets可能會(huì)接收高并發(fā)的請(qǐng)求，kafka默認(rèn)給其分配50個(gè)分區(qū)(可以通過(guò)
offsets.topic.num.partitions設(shè)置)，這樣可以通過(guò)加機(jī)器的方式抗大并發(fā)。
 

消費(fèi)者Rebalance機(jī)制

rebalance就是說(shuō)如果消費(fèi)組里的消費(fèi)者數(shù)量有變化或消費(fèi)的分區(qū)數(shù)有變化，kafka會(huì)重新分配消費(fèi)者消費(fèi)分區(qū)的關(guān)系。
比如consumer group中某個(gè)消費(fèi)者掛了，此時(shí)會(huì)自動(dòng)把分配給他的分區(qū)交給其他的消費(fèi)者，如果他又重啟了，那么又會(huì)
把一些分區(qū)重新交還給他。
注意：rebalance只針對(duì)subscribe這種不指定分區(qū)消費(fèi)的情況，如果通過(guò)assign這種消費(fèi)方式指定了分區(qū)，kafka不會(huì)進(jìn)
行rebanlance。
如下情況可能會(huì)觸發(fā)消費(fèi)者rebalance
1. 消費(fèi)組里的consumer增加或減少了
2. 動(dòng)態(tài)給topic增加了分區(qū)
3. 消費(fèi)組訂閱了更多的topic
rebalance過(guò)程中，消費(fèi)者無(wú)法從kafka消費(fèi)消息，這對(duì)kafka的TPS會(huì)有影響，如果kafka集群內(nèi)節(jié)點(diǎn)較多，比如數(shù)百
個(gè)，那重平衡可能會(huì)耗時(shí)極多，所以應(yīng)盡量避免在系統(tǒng)高峰期的重平衡發(fā)生。
Rebalance過(guò)程如下
當(dāng)有消費(fèi)者加入消費(fèi)組時(shí)，消費(fèi)者、消費(fèi)組及組協(xié)調(diào)器之間會(huì)經(jīng)歷以下幾個(gè)階段