Kafka Controller作業原理

一 導讀

Kafka是一個基于Scala開發的多磁區、多副本且基于ZooKeeper協調的分布式訊息系統，它以高吞吐、可水平擴展、支持流資料處理等多種特性而被廣泛使用，本文帶大家了解Kafka服務端核心模塊Kafka Controller，

二 Kafka簡介

Kafka起初由LinkedIn開發，后捐贈給Apache，主要用于日志收集、實時計算、MySQL Binlog日志分發等業務，Kafka集群由若干臺Broker、若干臺Producer、若干臺Consumer組成，此外Kafka集群需要依賴ZooKeeper集群做協調，完成元資料管理、優先副本選舉、管理Topic等作業，

三 Kafka Controller

3.1 選舉

Kafka Controller，其實就是一個Kafka集群中一臺Broker，它除了具有普通Broker的訊息發送、消費、同步功能之外，還需承擔一些額外的作業，Kafka使用公平競選的方式來確定Controller，最先在ZooKeeper成功創建臨時節點/controller的Broker會成為Controller，一般而言，Kafka集群中第一臺啟動的Broker會成為Controller，并將自身Broker編號等資訊寫入ZooKeeper臨時節點/controller，

其中version固定為1，brokerid表示成為Broker編號，timestamp表示成為Controller的時間戳，在任意時刻，Kafka集群中有且只有一臺Controller，每個Broker啟動的時候都會去嘗試讀取ZooKeeper臨時節點/controller的brokerid，如果讀取到brokerid不為-1，則表示已有Broker競選成為Controller，此時，當前Broker會放棄競選Controller，并對ZooKeeper臨時節點/controller注冊監聽器，當Controller因為某些情況導致Kafka行程停止，甚至所在的機器宕機，在規定的時間內沒有發送心跳到ZooKeeper，此時ZooKeeper會將臨時節點/controller洗掉，然后監聽器會通知各個Broker進行Controller選舉，第一個創建ZooKeeper臨時節點/controller被認為搶占成功，成為新的Controller，

3.2 腦裂

隨著服務的長時間運行，加上Controller對比普通Broker需承擔一部分額外的作業，Controller不可避免發生Full GC，導致服務停止，如果Full GC時間很長，甚至超過Broker與ZooKeeper設定的會話時間，此時ZooKeeper會認為Controller下線，并自動洗掉臨時節點/controller，進行新一輪的Controller選舉，

選舉產生新的Controller之后，ZooKeeper會通知到各個Broker更新記憶體中的ActiveControllerId，原先發生Full GC的Controller因為停止服務收不到Controller更新通知，在Full GC過后，原先的Controller認為自身服務正常，只是假死了一段時間，甚至還認為自己是Controller，此時Kafka集群就會有兩個Controller，這就是Controller腦裂問題，

為了解決Controller腦裂問題，ZooKeeper中還有一個與Controller有關的持久節點/controller_epoch，存放的是一個整形值，用于記錄Controller發生變更的次數，即記錄當前是第幾代Controller，也稱為紀元編號，Controller和普通Broker的請求都會攜帶controller_epoch，如果請求的controller_epoch小于ActiveControllerId，則認為這個請求是已經過期的Controller發送的無效請求，此時普通Broker，如果請求的controller_epoch大于記憶體中ActiveControllerId，那么說明已經有新的Controller產生，當前Broker需更新記憶體中的ActiveControllerId，Kafka通過controller_epoch來保證Controller的唯一性，同時防止Controller腦裂，

3.3 職責

擁有Controller身份的Broker，除了要像普通Broker那樣對外提供訊息的生產、消費、同步功能之外，還需多承擔以下幾份職責：

• 監聽磁區的變化，對ZooKeeper中的/admin/reassign_partitions節點注冊ReassignmentHandler，用來處理磁區重分配；對ZooKeeper中的/isr_change_notification節點注冊ChangeNotificetionHandler，用來處理ISR變更；對ZooKeeper中的/admin/preferred-relica-election節點注冊PreferredReplicaElectionHandler，用來處理優先副本的選舉，
• 監聽Topic的變化，對ZooKeeper中的/brokers/topics節點注冊TopicChangeHandler，用來處理Topic的增減變化；對ZooKeeper中的/admin/delete_topics節點注冊TopicDeleteHandler，用來處理洗掉Topic，
• 監聽Broker的變化，對ZooKeeper中的/brokers/ids節點注冊BrokerChangeHandler，用來處理Broker增減的變化，
• 從ZooKeeper中讀取當前所有與Topic、磁區以及Broker有關的資訊并進行相應的管理，對所有Topic對應的ZooKeeper中的/brokers/topics/<topic>節點添加PartitionModificationsHandler，用來監聽Topic中的磁區分配變化，
• 自動平衡優先副本，如果Broker啟動引數auto.leader.rebalance.enable設定為true，Controller會開啟定時任務，默認每五分鐘（啟動引數可配置）計算一次Kafka集群優先副本的不平衡率，如果超過10%（啟動引數可配置）會自動平衡磁區優先副本，避免因為磁區優先副本不均衡造成流量不均勻，進而影響個別Broker性能，

總結起來，Controller就是注冊一些處理監聽器，監聽事件做相應的處理，以及開啟一些定時任務，做相應的處理，二者的區別在于，監聽器能快速做出相應處理，比如Topic變更，這些事件的實時性要求很高，要求系統立即反應，優先副本不平衡，只是Broker讀寫流量不均衡，有一定程式的不平衡，對于Kafka集群來說是可以接受的，因此像磁區優先副本自動平衡就可以用定時任務來處理，

3.4 原理

Controller在選舉成功后會讀取ZooKeeper中各個節點的資訊來初始化ControllerContext，并管理這些ControllerContext，比如某個Topic增加若干個磁區，Controller在負責創建這些磁區的同時，還要更新ControllerContext，并且需要將這些變更資訊同步到普通Broker，不管是監聽器觸發的事件，還是定時任務觸發的事件，或者是其他事件都會讀取或者更新Controller中的ControllerContext，那么這樣就會涉及多執行緒同步，如果單純使用鎖機制來控制，那么整體性能會大打折扣，實際上，Kafka在0.10之前，都是通過鎖機制的方式來處理ControllerContext的多執行緒同步問題，在0.11后改用單執行緒基于事件佇列的模型，將每個事件都做一層封裝，然后按事件發生的先后順序暫存到LinkedBlockingQueue中，最后使用ControllerEventThread執行緒按照FIFO的原則來處理各個事件，這樣就可以不用鎖機制解決多執行緒同步問題，

變更Controller，通常有以下幾種方式，但是Controller上下線，需要付出的代價也比較大，需要關閉原Controller各種監聽器、定時任務、以及各種執行緒，并且新Controller上線同樣開啟以上資源，一般而言，除非Controller所在的機器IO、CPU、記憶體等資源不足，又或者頻繁發生長時間Full GC，否則不應該人為變更Controller，

• 手動更改ZooKeeper臨時/controller中brokerid，每個Broker會更新記憶體中的ActiveControllerId，
• 手動洗掉ZooKeeper臨時節點/controller，觸發一輪新的選舉，
• 停止當前Controller Kafka行程，此時ZooKeeper會認為Controller所在的Broker宕機，ZooKeeper同樣會洗掉臨時節點/controller，觸發一輪新的選舉，

3.5 監控

監控Controller個數尤為重要，需要為Controller建立監控指標，個數不為1的情況都需要特別留意，由于Kafka會在記憶體里面維護一些監控指標，并以JMX的方式對外提供，因此我們只需要在啟動Kafka時開放JMX埠，再定時獲取Controller對應的指標即可，

四 結語

本文通過講解Controller的選舉、腦裂、職責、原理和監控，加深對Controller的認識和了解，Controller是Kafka服務端的重要組成部分，除了要像普通Broker完成訊息的生產、消費、同步等功能之外，還需承擔管理磁區、副本狀態以及磁區重分配等一些額外的作業，