kafka Controller模組詳細解讀

集群元資料 controllercontext

controllerstats

shuttingdownbrokerids

epoch & epochzkversion

livebrokers

livebrokerepochs

alltopics

partitionassignments

partitionleadershipinfo

controller與broker之間的請求傳送 controllerchannelmanager

請求型別

leaderandisrrequest

stopreplicarequest

updatemetadatarequest

請求傳送執行緒（requestsendthread）

請求傳送類 controllerchannelmanager

startup（）

addbroker（）

sendrequest（）

**controllerchannelmanager

controllereventmanager(事件請求處理)

controllereventthread（單執行緒事件處理模型）

kafkacontroller

controller選舉場景

選舉controller

事實上，集群broker是不會與zookeeper直接互動去獲取元資料的。相反，它們總是與controller進行通訊，獲取和更新最新的集群資料。（而且，社群已經打算把zookeeper乾掉了）本文，我們講解關於kafka的乙個重要元件controller.

首先介紹下關於controller下的資料容器類。controllercontext

集群的元資料資訊都儲存在controllercontext類中，應該說，這個類是controller元件的資料容器類。

class controllercontext 
override def beforeinitializingsession(): unit = 
})//寫入startup事件到事件佇列
eventmanager.put(startup)
//啟動controllereventthread執行緒，開始處理事件佇列中的controllerevent
eventmanager.start()
}

startup方法註冊zk狀態變更***，用於監聽broker和zookeeper之間的會話是否過期，接著寫入startup事件到事件佇列，然後啟動controllereventthread執行緒，開始處理事件佇列裡的startup事件。

//處理startup事件 case object startup extends controllerevent

}

broker 偵測 /controller 節點消失時：

broker 檢測到 /controller 節點消失時，就意味著，此時整個集群中沒有 controller。因此，所有檢測到 /controller 節點消失的 broker，都會立即呼叫 elect 方法執行競選邏輯

/controller 節點資料變更：

如果 broker 之前是 controller，那麼該 broker 需要首先執行卸任操作，然後再嘗試競選；

如果 broker 之前不是 controller，那麼，該 broker 直接去競選新 controller。

//卸任邏輯
private def oncontrollerresignation()

//controller選舉
private def elect(): unit = 
try  successfully elected as the controller")
//將建立到controller節點的broker指定為新的controller
activecontrollerid = config.brokerid
//執行當選controller的後續邏輯
oncontrolle***ilover()
} catch ")
else
warn("a controller has been elected but just resigned, this will result in another round of election")
case e2: throwable =>
error(s"error while electing or becoming controller on broker $", e2)
triggercontrollermove()
}}

該方法，首先檢查controller是否已經選出來了，集群中的所有broker都要執行這些邏輯，所以，非常有可能出現某些broker在執行elect方法時，controller已經被選出來了，如果選出來，自然不需要在做什麼，如果沒有，**會嘗試建立/controller節點去搶註controller.

一旦搶註成功，就會呼叫oncontrolle***ilover方法，執行選舉成功後的操作，比如註冊各類zk***，刪除日誌路徑變更，isr副本變更通知，啟動controller通道管理器，啟動副本狀態機和分割槽狀態機等。

//當選controller的後續邏輯
private def oncontrolle***ilover() 
if (config.tokenauthenabled) 
}

kafka Controller模組詳細解讀

MySQL TIMESTAMP 時間戳詳細解釋

Spring 註解 Qualifier 詳細解析

Linux etc shadow 超詳細內容解析

kafka Controller模組 詳細解讀

MySQL TIMESTAMP 時間戳 詳細解釋

Spring 註解 Qualifier 詳細解析

Linux etc shadow 超詳細內容解析

相關推薦

kafka Controller模組詳細解讀

MySQL TIMESTAMP 時間戳詳細解釋