Kafka控制器選舉流程剖析
1.概述
平時在使用Kafka的時候，可能關注的更多的是Kafka系統層面的。今天來給大家剖析一下Kafka的控制器，瞭解一下Kafka控制器的選舉流程。
2.內容
Kafka控制器，其實就是一個Kafka系統的Broker。它除了具有一般Broker的功能之外，還具有選舉主題分割槽Leader節點的功能。在啟動Kafka系統時，其中一個Broker會被選舉為控制器，負責管理主題分割槽和副本狀態，還會執行分割槽重新分配的管理任務。
如果在Kafka系統執行過程中，當前的控制器出現故障導致不可用，那麼Kafka系統會從其他正常執行的Broker中重新選舉出新的控制器。
2.1 控制器啟動順序
在Kafka叢集中，每個Broker在啟動時會例項化一個KafkaController類。該類會執行一系列業務邏輯，選舉出主題分割槽的Leader節點，步驟如下：
- 第一個啟動的代理節點，會在Zookeeper系統裡面建立一個臨時節點/controller，並寫入該節點的註冊資訊，使該節點成為控制器；
- 其他的代理節點陸續啟動時，也會嘗試在Zookeeper系統中建立/controller節點，但是由於/controller節點已經存在，所以會丟擲“建立/controller節點失敗異常”的資訊。建立失敗的代理節點會根據返回的結果，判斷出在Kafka叢集中已經有一個控制器被成功建立了，所以放棄建立/controller節點，這樣就確保了Kafka叢集控制器的唯一性；
- 其他的代理節點，會在控制器上註冊相應的監聽器，各個監聽器負責監聽各自代理節點的狀態變化。當監聽到節點狀態發生變化時，會觸發相應的監聽函式進行處理。
2.2 如何檢視控制器優先順序 ?
控制器建立的優先順序是按照Kafka系統代理節點成功啟動的順序來建立的。使用者可以通過改變Kafka系統代理節點的啟動順序，來檢視控制器的建立優先順序。之後，可以在Zookeeper系統中檢視/controller臨時節點的內容，例如：
# 進入Zookeeper叢集 [hadoop@dn1 bin]$zkCli.sh -server dn1:2181 # 執行檢視命令 [zk: dn1:2181(CONNECTED) 1] get /controller
成功執行命令後，可以看到代理節點0（即dn1節點）上成功建立了控制器，如下圖所示：
當前啟動順序為：dn1、dn2、dn3，修改啟動順序為：dn3、dn1、dn2。再次檢視Zookeeper系統中執行“get /controller”命令，輸出結果如下圖所示：
2.3 切換控制器所屬的代理節點
當控制器被關閉或者與Zookeeper系統斷開連線時，Zookeeper系統上的臨時節點就會被清除。Kafka叢集中的監聽器會接收到變更通知，各個代理節點會嘗試到Zookeeper系統中建立一個控制器的臨時節點。第一個成功在Zookeeper系統中建立的代理節點，將會成為新的控制器。每個新選舉出來的控制器，會在Zookeeper系統中獲取一個遞增的controller_epoch值。
3.主題分割槽Leader節點的選舉過程
選舉控制器的核心思路是：各個代理節點公平競爭搶佔Zookeeper系統中建立/controller臨時節點，最先建立成功的代理節點會成為控制器，並擁有選舉主題分割槽Leader節點的功能。選舉流程如下圖所示：
當Kafka系統例項化KafkaController類時，主題分割槽Leader節點的選舉流程便會開始。其中涉及的核心類包含KafkaController、ZookeeperLeaderElector、LeaderChangeListener、SessionExpirationListener。
- KafkaController：在例項化ZookeeperLeaderElector類時，分別設定了兩個關鍵的回撥函式，即onControllerFailover和onControllerResignation；
- ZookeeperLeaderElector：實現主題分割槽的Leader節點選舉功能，但是它並不會處理“代理節點與Zookeeper系統之間出現的會話超時”這種情況，它主要負責建立元資料儲存路徑、例項化變更監聽器等，並通過訂閱資料變更監聽器來實時監聽資料的變化，進而開始執行選舉Leader的邏輯；
- LeaderChangeListener：如果節點資料傳送變化，則Kafka系統中的其他代理節點可能已經成為Leader，接著Kafka控制器會呼叫onResigningAsLeader函式。當Kafka代理節點宕機或者被人為誤刪除時，則處於該節點上的Leader會被重新選舉，通過呼叫onResigningAsLeader函式重新選擇其他正常執行的代理節點成為新的Leader；
- SessionExpirationListener：當Kafka系統的代理節點和Zookeeper系統建立連線後，SessionExpirationListener中的handleNewSession函式會被呼叫，對於Zookeeper系統中會話過期的連線，會先進行一次判斷。
4.註冊分割槽和副本狀態機
Kafka系統的控制器主要負責管理主題、分割槽和副本。 Kafka系統在操作主題、分割槽和副本時，控制器會在Zookeeper系統的/brokers/topics節點，以及其子節點路徑上註冊一系列的監聽器。 使用Kafka應用介面或者是Kafka系統指令碼建立一個主題時，服務端會將建立後的結果返回給客戶端。當客戶端收到建立成功的提示時，其實服務端並沒有實際建立主題，而只是在Zookeeper系統的/brokers/topics節點中建立了該主題對應的子節點名稱。
代理節點呼叫onBecomingLeader()函式實際上呼叫的是onControllerFailover()函式，所以在控制器呼叫onControllerFailover()函式時，會在初始化階段分別建立分割槽狀態機和副本狀態機。程式碼如下所示：
def onControllerFailover() {
if(isRunning) {
info("Broker %d starting become controller state
transition".format(config.brokerId))
readControllerEpochFromZookeeper()
incrementControllerEpoch(zkUtils.zkClient)
// 在/brokers/topics節點註冊監聽器
registerReassignedPartitionsListener()
registerIsrChangeNotificationListener()
registerPreferredReplicaElectionListener()
partitionStateMachine.registerListeners()// 註冊分割槽狀態機
replicaStateMachine.registerListeners()// 註冊副本狀態機
initializeControllerContext()
// 在控制器初始化之後，在狀態機啟動之前，需要傳送更新元資料請求
sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)
replicaStateMachine.startup()// 啟動副本狀態機
partitionStateMachine.startup()// 啟動分割槽狀態機
// 在自動故障轉移中為所有主題註冊分割槽更改監聽器
controllerContext.allTopics.foreach(topic => partitionStateMachine.
registerPartitionChangeListener(topic))
info("Broker %d is ready to serve as the new controller with epoch %d".
format(config.brokerId, epoch))
maybeTriggerPartitionReassignment()
maybeTriggerPreferredReplicaElection()
if (config.autoLeaderRebalanceEnable) {
info("starting the partition rebalance scheduler")
autoRebalanceScheduler.startup()
autoRebalanceScheduler.schedule("partition-rebalance-thread",
checkAndTriggerPartitionRebalance,
5,
config.leaderImbalanceCheckIntervalSeconds.toLong,
TimeUnit.SECONDS)
}
deleteTopicManager.start()
}
else
info("Controller has been shut down, aborting startup/failover")
}
主題的分割槽狀態機通過registerListeners()函式，在Zookeeper系統中的/brokers/topics節點上註冊了TopicChangeListener和DeleteTopicListener兩個監聽器。建立一個主題時，主題資訊、主題分割槽和副本會被寫到Zookeeper系統的/brokers/topics節點中，這就會觸發分割槽和副本狀態機註冊監聽器。
5.總結
Kafka系統整體來說，除錯還算方便。下載Kafka原始碼，匯入到IDE中，就可以啟動整個Kafka系統了，可以通過DEBUG的方式來親自了解控制器的執行流程。
6.結束語
這篇部落格就和大家分享到這裡，如果大家在研究學習的過程當中有什麼問題，可以加群進行討論或傳送郵件給我，我會盡我所能為您解答，與君共勉！
另外，博主出書了《Hadoop大資料探勘從入門到進階實戰》，喜歡的朋友或同學， 可以在公告欄那裡點選購買連結購買博主的書進行學習，在此感謝大家的支援。