Kafka核心總控制器Controller

在Kafka集群中會有一個或者多個broker，其中有一個broker會被選舉為控制器(Kafka Controller)，它負(fù)責(zé)管理整個集群中所有分區(qū)和副本的狀態(tài)。

• 當(dāng)某個分區(qū)的leader副本出現(xiàn)故障時，由控制器負(fù)責(zé)為該分區(qū)選舉新的leader副本。
• 當(dāng)檢測到某個分區(qū)的ISR集合發(fā)生變化時，由控制器負(fù)責(zé)通知所有broker更新其元數(shù)據(jù)信息。
• 當(dāng)使用kafka-topics.sh腳本為某個topic增加分區(qū)數(shù)量時，同樣還是由控制器負(fù)責(zé)讓新分區(qū)被其他節(jié)點感知到。

Controller選舉機制

在kafka集群啟動的時候，會自動選舉一臺broker作為controller來管理整個集群，選舉的過程是集群中每個broker都會嘗試在zookeeper上創(chuàng)建一個 /controller 臨時節(jié)點，zookeeper會保證有且僅有一個broker能創(chuàng)建成功，這個broker就會成為集群的總控器controller。 當(dāng)這個controller角色的broker宕機了，此時zookeeper臨時節(jié)點會消失，集群里其他broker會一直監(jiān)聽這個臨時節(jié)點，發(fā)現(xiàn)臨時節(jié)點消失了，就競爭再次創(chuàng)建臨時節(jié)點，就是我們上面說的選舉機制，zookeeper又會保證有一個broker成為新的controller。

• 監(jiān)聽broker相關(guān)的變化。為Zookeeper中的/brokers/ids/節(jié)點添加BrokerChangeListener，用來處理broker增減的變化。
• 監(jiān)聽topic相關(guān)的變化。為Zookeeper中的/brokers/topics節(jié)點添加TopicChangeListener，用來處理topic增減的變化;為Zookeeper中的/admin/delete_topics節(jié)點添加TopicDeletionListener，用來處理刪除topic的動作。

從Zookeeper中讀取獲取當(dāng)前所有與topic、partition以及broker有關(guān)的信息并進行相應(yīng)的管理。對于所有topic所對應(yīng)的Zookeeper中的/brokers/topics/[topic]節(jié)點添加PartitionModificationsListener，用來監(jiān)聽topic中的分區(qū)分配變化。

更新集群的元數(shù)據(jù)信息，同步到其他普通的broker節(jié)點中。

Partition副本選舉Leader機制

controller感知到分區(qū)leader所在的broker掛了(controller監(jiān)聽了很多zk節(jié)點可以感知到broker存活)，controller會從每個parititon的 replicas 副本列表中取出第一個broker作為leader，當(dāng)然這個broker需要也同時在ISR列表里。

消費者消費消息的offset記錄機制

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

消費者Rebalance機制

消費者rebalance就是說如果consumer group中某個消費者掛了，此時會自動把分配給他的分區(qū)交給其他的消費者，如果他又重啟了，那么又會把一些分區(qū)重新交還給他。

注意：rebalance只針對subscribe這種不指定分區(qū)消費的情況，如果通過assign這種消費方式指定了分區(qū)，kafka不會進行rebanlance。

如下情況可能會觸發(fā)消費者rebalance

• consumer所在服務(wù)重啟或宕機了
• 動態(tài)給topic增加了分區(qū)
• 消費組訂閱了更多的topic

Rebalance過程

第一階段：選擇組協(xié)調(diào)器

組協(xié)調(diào)器GroupCoordinator：每個consumer group都會選擇一個broker作為自己的組協(xié)調(diào)器coordinator，負(fù)責(zé)監(jiān)控這個消費組里的所有消費者的心跳，以及判斷是否宕機，然后開啟消費者rebalance。

consumer group中的每個consumer啟動時會向kafka集群中的某個節(jié)點發(fā)送 FindCoordinatorRequest 請求來查找對應(yīng)的組協(xié)調(diào)器GroupCoordinator，并跟其建立網(wǎng)絡(luò)連接。

組協(xié)調(diào)器選擇方式：通過如下公式可以選出consumer消費的offset要提交到__consumer_offsets的哪個分區(qū)，這個分區(qū)leader對應(yīng)的broker就是這個consumer group的coordinator

公式：hash(consumer group id) % __consumer_offsets主題的分區(qū)數(shù)

第二階段：加入消費組JOIN GROUP

在成功找到消費組所對應(yīng)的 GroupCoordinator 之后就進入加入消費組的階段，在此階段的消費者會向 GroupCoordinator 發(fā)送 JoinGroupRequest 請求，并處理響應(yīng)。然后GroupCoordinator 從一個consumer group中選擇第一個加入group的consumer作為leader(消費組協(xié)調(diào)器)，把consumer group情況發(fā)送給這個leader，接著這個leader會負(fù)責(zé)制定分區(qū)方案。

第三階段( SYNC GROUP)

consumer leader通過給GroupCoordinator發(fā)送SyncGroupRequest，接著GroupCoordinator就把分區(qū)方案下發(fā)給各個consumer，他們會根據(jù)指定分區(qū)的leader broker進行網(wǎng)絡(luò)連接以及消息消費。

消費者Rebalance分區(qū)分配策略

主要有三種rebalance的策略：range、round-robin、sticky。

Kafka 提供了消費者客戶端參數(shù)partition.assignment.strategy 來設(shè)置消費者與訂閱主題之間的分區(qū)分配策略。默認(rèn)情況為range分配策略。

假設(shè)一個主題有10個分區(qū)(0-9)，現(xiàn)在有三個consumer消費：

1. range策略：按照分區(qū)序號排序，假設(shè) n=分區(qū)數(shù)/消費者數(shù)量 = 3， m=分區(qū)數(shù)%消費者數(shù)量 = 1，那么前 m 個消費者每個分配 n+1 個分區(qū)，后面的(消費者數(shù)量-m )個消費者每個分配 n 個分區(qū)。比如分區(qū)0~3給一個consumer，分區(qū)4~6給一個consumer，分區(qū)7~9給一個consumer。
2. round-robin策略：輪詢分配，比如分區(qū)0、3、6、9給一個consumer，分區(qū)1、4、7給一個consumer，分區(qū)2、5、8給一個consumer
3. sticky策略：rebalance的時候，需要保證如下兩個原則。

• 分區(qū)的分配要盡可能均勻 。
• 分區(qū)的分配盡可能與上次分配的保持相同。

當(dāng)兩者發(fā)生沖突時，第一個目標(biāo)優(yōu)先于第二個目標(biāo) 。這樣可以最大程度維持原來的分區(qū)分配的策略。

比如對于第一種range情況的分配，如果第三個consumer掛了，那么重新用sticky策略分配的結(jié)果如下：

• consumer1除了原有的0~3，會再分配一個7
• consumer2除了原有的4~6，會再分配8和9

producer發(fā)布消息機制剖析

1、寫入方式

producer 采用 push 模式將消息發(fā)布到 broker，每條消息都被 append 到 patition 中，屬于順序?qū)懘疟P(順序?qū)懘疟P效率比隨機寫內(nèi)存要高，保障 kafka 吞吐率)。

2、消息路由

producer 發(fā)送消息到 broker 時，會根據(jù)分區(qū)算法選擇將其存儲到哪一個 partition。其路由機制為：1. 指定了 patition，則直接使用; 2. 未指定 patition 但指定 key，通過對 key 的 value 進行hash 選出一個 patition 3. patition 和 key 都未指定，使用輪詢選出一個 patition。

3、寫入流程  

• producer 先從 zookeeper 的 "/brokers/.../state" 節(jié)點找到該 partition 的 leader
• producer 將消息發(fā)送給該 leader
• leader 將消息寫入本地 log
• followers 從 leader pull 消息，寫入本地 log 后 向leader 發(fā)送 ACK
• leader 收到所有 ISR 中的 replica 的 ACK 后，增加 HW(high watermark，最后 commit 的 offset) 并向 producer 發(fā)送 ACK

HW與LEO詳解

HW俗稱高水位，HighWatermark的縮寫，取一個partition對應(yīng)的ISR中最小的LEO(log-end-offset)作為HW，consumer最多只能消費到HW所在的位置。另外每個replica都有HW,leader和follower各自負(fù)責(zé)更新自己的HW的狀態(tài)。對于leader新寫入的消息，consumer不能立刻消費，leader會等待該消息被所有ISR中的replicas同步后更新HW，此時消息才能被consumer消費。這樣就保證了如果leader所在的broker失效，該消息仍然可以從新選舉的leader中獲取。對于來自內(nèi)部broker的讀取請求，沒有HW的限制。

舉例當(dāng)producer生產(chǎn)消息至broker后，ISR以及HW和LEO的流轉(zhuǎn)過程 

由此可見，Kafka的復(fù)制機制既不是完全的同步復(fù)制，也不是單純的異步復(fù)制。事實上，同步復(fù)制要求所有能工作的follower都復(fù)制完，這條消息才會被commit，這種復(fù)制方式極大的影響了吞吐率。而異步復(fù)制方式下，follower異步的從leader復(fù)制數(shù)據(jù)，數(shù)據(jù)只要被leader寫入log就被認(rèn)為已經(jīng)commit，這種情況下如果follower都還沒有復(fù)制完，落后于leader時，突然leader宕機，則會丟失數(shù)據(jù)。而Kafka的這種使用ISR的方式則很好的均衡了確保數(shù)據(jù)不丟失以及吞吐率。

結(jié)合HW和LEO看下 acks=1的情況