Kafka中的再均衡

在《Kafka消費者的使用和原理》中已經提到過「再均衡」的概念，咱們先回顧下，一個主題能夠有多個分區，而訂閱該主題的消費組中能夠有多個消費者。每個分區只能被消費組中的一個消費者消費，可認爲每一個分區的消費權只屬於消費組中的一個消費者。可是世界是變化的，例如消費者會宕機，還有新的消費者會加入，而爲了應對這些變化，讓分區所屬權的分配合理，這都須要對分區所屬權進行調整，也就是所謂的「再均衡」。本文將對再均衡的相關知識進行詳細敘述。

觸發時機

首先，咱們須要瞭解什麼狀況下會觸發再均衡，在前文已經提到了消費者數量的變化，是須要再均衡的。在使用Kafka時，除了消費者數量可能會變化，分區數量也一樣可能變化，咱們能夠人爲的對分區數量進行修改，可是Kafka只容許增長分區，因此咱們只能把分區數量調大，不能調小，不然會收到InvalidPartitionException異常。關於爲何不能減小分區，可參考下面的回答：

按Kafka現有的代碼邏輯，此功能是徹底能夠實現的，不過也會使得代碼的複雜度急劇增大。實現此功能須要考慮的因素不少，好比刪除掉的分區中的消息該做何處理？若是隨着分區一塊兒消失則消息的可靠性得不到保障；若是須要保留則又須要考慮如何保留。直接存儲到現有分區的尾部，消息的時間戳就不會遞增，如此對於Spark、Flink這類須要消息時間戳（事件時間）的組件將會受到影響；若是分散插入到現有的分區中，那麼在消息量很大的時候，內部的數據複製會佔用很大的資源，並且在複製期間，此主題的可用性又如何獲得保障？與此同時，順序性問題、事務性問題、以及分區和副本的狀態機切換問題都是不得不面對的。反觀這個功能的收益點倒是很低，若是真的須要實現此類的功能，徹底能夠從新建立一個分區數較小的主題，而後將現有主題中的消息按照既定的邏輯複製過去便可。

簡單來講，就是作這個功能須要考慮不少因素，這樣會把代碼弄的很複雜，而收益卻很低，並且存在替代方案來實現該效果，建立一個分區數小的主題，再把當前主題遷移過去。

除了消費者、分區數量的變化，還有一種狀況，也須要進行再均衡。當消費者訂閱主題時使用的是正則表達式，例如「test.*」，表示訂閱全部以test開頭的主題，當有新的以test開頭的主題被建立時，則須要經過再均衡將該主題的分區分配給消費者。

再均衡的三種觸發時機，咱們已經清楚了，下面咱們看下再均衡是如何實現的。

協調者

再均衡，將分區所屬權分配給消費者。所以須要和全部消費者通訊，這就須要引進一個協調者的概念，由協調者爲消費組服務，爲消費者們作好協調工做。在Kafka中，每一臺Broker上都有一個協調者組件，負責組成員管理、再均衡和提交位移管理等工做。若是有N臺Broker，那就有N個協調者組件，而一個消費組只需一個協調者進行服務，那該由哪一個Broker爲其服務？肯定Broker須要兩步：

1. 計算分區號
   partition = Math.abs(groupID.hashCode() % offsetsTopicPartitionCount)

根據groupID的哈希值，取餘offsetsTopicPartitionCount（內部主題consumer_offsets的分區數，默認50）的絕對值，其意思就是把消費組哈希散列到內部主題__consumer_offsets的一個分區上。肯定協調者爲何要和內部主題扯上關係。這就跟協調者的做用有關了。協調者不只是負責組成員管理和再均衡，在協調者中還須要負責處理消費者的偏移量提交，而偏移量提交則正是提交到consumer_offsets的一個分區上。因此這裏須要取餘offsetsTopicPartitionCount來肯定偏移量提交的分區。

1. 找出分區Leader副本所在的Broker

肯定了分區就簡單了，分區Leader副本所在的Broker上的協調者，就是咱們要找的。

這個算法一般用於幫助定位問題。當一個消費組出現問題時，咱們能夠先肯定協調者的Broker，而後查看Broker端的日誌來定位問題。

交互方式

協調者，咱們肯定了。那協調者和消費者之間是如何交互的？協調者如何掌握消費者的狀態，又如何通知再均衡。這裏使用了心跳機制。在消費者端有一個專門的心跳線程負責以heartbeat.interval.ms的間隔頻率發送心跳給協調者，告訴協調者本身還活着。同時協調者會返回一個響應。而當須要開始再均衡時，協調者則會在響應中加入REBALANCE_IN_PROGRESS，當消費者收到響應時，便能知道再均衡要開始了。

因爲再平衡的開始依賴於心跳的響應，因此heartbeat.interval.ms除了決定心跳的頻率，也決定了再均衡的通知頻率。
如今咱們再從新看下，觸發再均衡的時機，前面說到有三種狀況觸發再均衡，分別是消費者數量的增長或減小、分區數的增長和新建立主題，其中消費者數量增長、分區數增長和新建立主題，這都是必須是人爲操做，算是計劃內的再均衡。而消費者數量減小則除了是人爲操做，也可能由於其餘緣由致使，屬於計劃以外的再均衡，這是咱們須要關心的，畢竟再均衡的開銷仍是很大的，全部消費者都會中止工做，因此咱們應儘可能避免沒必要要的再均衡。下面咱們看下影響消費者數量減小的參數有哪些：

1. session.timeout.ms：Broker端參數，消費者的存活時間，默認10秒，若是在這段時間內，協調者沒收到任何心跳，則認爲該消費者已崩潰離組；
2. heartbeat.interval.ms：消費者端參數，發送心跳的頻率，默認3秒；
3. max.poll.interval.ms：消費者端參數，兩次調用poll的最大時間間隔，默認5分鐘，若是5分鐘內沒法消費完，則會主動離組。

能夠看出session.timeout.ms和heartbeat.interval.ms是相關的，這裏給出一個建議參考的公式：

session.timeout.ms ≥ 3 * heartbeat.interval.ms

爲儘可能避免由於偶發的網絡緣由，心跳沒法到達協調者，在超時以前，應至少能發送3輪心跳。再給出一個經驗值的設置：session.timeout.ms=6s，heartbeat.interval.ms=2s。

max.poll.interval.ms的設置，則主要和下游處理時間有關，例以下游處理時間須要6分鐘，那按默認值是不合理的，消費者會頻繁主動離組。因此須要把值設置的比下游處理時間大一點，避免沒必要要的再均衡。

這一小節主要講了協調者如何通知消費者開始再均衡，以及如何設置參數避免沒必要要的再均衡，下面咱們看下再均衡的流程是怎麼樣的。

流程

1. 當消費者收到協調者的再均衡開始通知時，須要當即提交偏移量；
2. 消費者在收到提交偏移量成功的響應後，再發送JoinGroup請求，從新申請加入組，請求中會含有訂閱的主題信息；
3. 當協調者收到第一個JoinGroup請求時，會把發出請求的消費者指定爲Leader消費者，同時等待rebalance.timeout.ms，在收集其餘消費者的JoinGroup請求中的訂閱信息後，將訂閱信息放在JoinGroup響應中發送給Leader消費者，並告知他成爲了Leader，同時也會發送成功入組的JoinGroup響應給其餘消費者；
4. Leader消費者收到JoinGroup響應後，根據消費者的訂閱信息制定分配方案，把方案放在SyncGroup請求中，發送給協調者。普通消費者在收到響應後，則直接發送SyncGroup請求，等待Leader的分配方案；
5. 協調者收到分配方案後，再經過SyncGroup響應把分配方案發給全部消費組。
6. 當全部消費者收到分配方案後，就意味着再均衡的結束，能夠正常開始消費工做了。