1. 程式人生 > >Kafka消費組(consumer group)

Kafka消費組(consumer group)

在開始之前,我想花一點時間先來明確一些概念和術語,這會極大地方便我們下面的討論。另外請原諒這文章有點長,畢竟要討論的東西很多,雖然已然刪除了很多太過細節的東西。

一、 誤區澄清與概念明確

1 Kafka的版本

很多人在Kafka中國社群(替群主做個宣傳,QQ號:162272557)提問時的開頭經常是這樣的:“我使用的kafka版本是2.10/2.11, 現在碰到一個奇怪的問題。。。。” 無意冒犯,但這裡的2.10/2.11不是kafka的版本,而是編譯kafka的Scala版本。Kafka的server端程式碼是由Scala語言編寫的,目前Scala主流的3個版本分別是2.10、2.11和2.12。實際上Kafka現在每個PULL request都已經自動增加了這三個版本的檢查。下圖是我的一個PULL request,可以看到這個fix會同時使用3個scala版本做編譯檢查:

目前廣泛使用kafka的版本應該是這三個大版本:0.8.x, 0.9.x和0.10.* 。 這三個版本對於consumer和consumer group來說都有很大的變化,我們後面會詳談。

2 新版本 VS 老版本

“我的kafkaoffsetmonitor為什麼無法監控到offset了?”——這是我在Kafka中國社群見到最多的問題,沒有之一!實際上,Kafka 0.9開始提供了新版本的consumer及consumer group,位移的管理與儲存機制發生了很大的變化——新版本consumer預設將不再儲存位移到zookeeper中,而目前kafkaoffsetmonitor還沒有應對這種變化(雖然已經有很多人在要求他們改了,詳見

https://github.com/quantifind/KafkaOffsetMonitor/issues/79),所以很有可能是因為你使用了新版本的consumer才無法看到的。關於新舊版本,這裡統一說明一下:kafka0.9以前的consumer是使用Scala編寫的,包名結構是kafka.consumer.*,分為high-level consumer和low-level consumer兩種。我們熟知的ConsumerConnector、ZookeeperConsumerConnector以及SimpleConsumer就是這個版本提供的;自0.9版本開始,Kafka提供了java版本的consumer,包名結構是o.a.k.clients.consumer.*,熟知的類包括KafkaConsumer和ConsumerRecord等。新版本的consumer可以單獨部署,不再需要依賴server端的程式碼。

二、消費者組 (Consumer Group)

1 什麼是消費者組

其實對於這些基本概念的普及,網上資料實在太多了。我本不應該再畫蛇添足了,但為了本文的完整性,我還是要花一些篇幅來重談consumer group,至少可以說說我的理解。值得一提的是,由於我們今天基本上只探討consumer group,對於單獨的消費者不做過多討論。

什麼是consumer group? 一言以蔽之,consumer group是kafka提供的可擴充套件且具有容錯性的消費者機制。既然是一個組,那麼組內必然可以有多個消費者或消費者例項(consumer instance),它們共享一個公共的ID,即group ID。組內的所有消費者協調在一起來消費訂閱主題(subscribed topics)的所有分割槽(partition)。當然,每個分割槽只能由同一個消費組內的一個consumer來消費。(網上文章中說到此處各種炫目多彩的圖就會緊跟著丟擲來,我這裡就不畫了,請原諒)。個人認為,理解consumer group記住下面這三個特性就好了:

  • consumer group下可以有一個或多個consumer instance,consumer instance可以是一個程序,也可以是一個執行緒
  • group.id是一個字串,唯一標識一個consumer group
  • consumer group下訂閱的topic下的每個分割槽只能分配給某個group下的一個consumer(當然該分割槽還可以被分配給其他group)

2 消費者位置(consumer position) 

消費者在消費的過程中需要記錄自己消費了多少資料,即消費位置資訊。在Kafka中這個位置資訊有個專門的術語:位移(offset)。很多訊息引擎都把這部分資訊儲存在伺服器端(broker端)。這樣做的好處當然是實現簡單,但會有三個主要的問題:1. broker從此變成有狀態的,會影響伸縮性;2. 需要引入應答機制(acknowledgement)來確認消費成功。3. 由於要儲存很多consumer的offset資訊,必然引入複雜的資料結構,造成資源浪費。而Kafka選擇了不同的方式:每個consumer group儲存自己的位移資訊,那麼只需要簡單的一個整數表示位置就夠了;同時可以引入checkpoint機制定期持久化,簡化了應答機制的實現。

3 位移管理(offset management)

3.1 自動VS手動

Kafka預設是定期幫你自動提交位移的(enable.auto.commit = true),你當然可以選擇手動提交位移實現自己控制。另外kafka會定期把group消費情況儲存起來,做成一個offset map,如下圖所示:

上圖中表明瞭test-group這個組當前的消費情況。

3.2 位移提交

老版本的位移是提交到zookeeper中的,圖就不畫了,總之目錄結構是:/consumers/<group.id>/offsets/<topic>/<partitionId>,但是zookeeper其實並不適合進行大批量的讀寫操作,尤其是寫操作。因此kafka提供了另一種解決方案:增加__consumeroffsets topic,將offset資訊寫入這個topic,擺脫對zookeeper的依賴(指儲存offset這件事情)。__consumer_offsets中的訊息儲存了每個consumer group某一時刻提交的offset資訊。依然以上圖中的consumer group為例,格式大概如下:

__consumers_offsets topic配置了compact策略,使得它總是能夠儲存最新的位移資訊,既控制了該topic總體的日誌容量,也能實現儲存最新offset的目的。compact的具體原理請參見:Log Compaction

4 Rebalance

4.1 什麼是rebalance?

rebalance本質上是一種協議,規定了一個consumer group下的所有consumer如何達成一致來分配訂閱topic的每個分割槽。比如某個group下有20個consumer,它訂閱了一個具有100個分割槽的topic。正常情況下,Kafka平均會為每個consumer分配5個分割槽。這個分配的過程就叫rebalance。

4.2 什麼時候rebalance?

這也是經常被提及的一個問題。rebalance的觸發條件有三種:

  • 組成員發生變更(新consumer加入組、已有consumer主動離開組或已有consumer崩潰了——這兩者的區別後面會談到)
  • 訂閱主題數發生變更——這當然是可能的,如果你使用了正則表示式的方式進行訂閱,那麼新建匹配正則表示式的topic就會觸發rebalance
  • 訂閱主題的分割槽數發生變更

4.3 如何進行組內分割槽分配?

之前提到了group下的所有consumer都會協調在一起共同參與分配,這是如何完成的?Kafka新版本consumer預設提供了兩種分配策略:range和round-robin。當然Kafka採用了可插拔式的分配策略,你可以建立自己的分配器以實現不同的分配策略。實際上,由於目前range和round-robin兩種分配器都有一些弊端,Kafka社群已經提出第三種分配器來實現更加公平的分配策略,只是目前還在開發中。我們這裡只需要知道consumer group預設已經幫我們把訂閱topic的分割槽分配工作做好了就行了。

簡單舉個例子,假設目前某個consumer group下有兩個consumer: A和B,當第三個成員加入時,kafka會觸發rebalance並根據預設的分配策略重新為A、B和C分配分割槽,如下圖所示:

4.4 誰來執行rebalance和consumer group管理?

Kafka提供了一個角色:coordinator來執行對於consumer group的管理。坦率說kafka對於coordinator的設計與修改是一個很長的故事。最新版本的coordinator也與最初的設計有了很大的不同。這裡我只想提及兩次比較大的改變。

首先是0.8版本的coordinator,那時候的coordinator是依賴zookeeper來實現對於consumer group的管理的。Coordinator監聽zookeeper的/consumers/<group>/ids的子節點變化以及/brokers/topics/<topic>資料變化來判斷是否需要進行rebalance。group下的每個consumer都自己決定要消費哪些分割槽,並把自己的決定搶先在zookeeper中的/consumers/<group>/owners/<topic>/<partition>下注冊。很明顯,這種方案要依賴於zookeeper的幫助,而且每個consumer是單獨做決定的,沒有那種“大家屬於一個組,要協商做事情”的精神。

基於這些潛在的弊端,0.9版本的kafka改進了coordinator的設計,提出了group coordinator——每個consumer group都會被分配一個這樣的coordinator用於組管理和位移管理。這個group coordinator比原來承擔了更多的責任,比如組成員管理、位移提交保護機制等。當新版本consumer group的第一個consumer啟動的時候,它會去和kafka server確定誰是它們組的coordinator。之後該group內的所有成員都會和該coordinator進行協調通訊。顯而易見,這種coordinator設計不再需要zookeeper了,效能上可以得到很大的提升。後面的所有部分我們都將討論最新版本的coordinator設計。

4.5 如何確定coordinator?

上面簡單討論了新版coordinator的設計,那麼consumer group如何確定自己的coordinator是誰呢? 簡單來說分為兩步:

  • 確定consumer group位移資訊寫入__consumers_offsets的哪個分割槽。具體計算公式:
    • __consumers_offsets partition# = Math.abs(groupId.hashCode() % groupMetadataTopicPartitionCount)   注意:groupMetadataTopicPartitionCount由offsets.topic.num.partitions指定,預設是50個分割槽。
  • 該分割槽leader所在的broker就是被選定的coordinator

4.6 Rebalance Generation

JVM GC的分代收集就是這個詞(嚴格來說是generational),我這裡把它翻譯成“屆”好了,它表示了rebalance之後的一屆成員,主要是用於保護consumer group,隔離無效offset提交的。比如上一屆的consumer成員是無法提交位移到新一屆的consumer group中。我們有時候可以看到ILLEGAL_GENERATION的錯誤,就是kafka在抱怨這件事情。每次group進行rebalance之後,generation號都會加1,表示group進入到了一個新的版本,如下圖所示: Generation 1時group有3個成員,隨後成員2退出組,coordinator觸發rebalance,consumer group進入Generation 2,之後成員4加入,再次觸發rebalance,group進入Generation 3.

4.7 協議(protocol)

前面說過了, rebalance本質上是一組協議。group與coordinator共同使用它來完成group的rebalance。目前kafka提供了5個協議來處理與consumer group coordination相關的問題:

  • Heartbeat請求:consumer需要定期給coordinator傳送心跳來表明自己還活著
  • LeaveGroup請求:主動告訴coordinator我要離開consumer group
  • SyncGroup請求:group leader把分配方案告訴組內所有成員
  • JoinGroup請求:成員請求加入組
  • DescribeGroup請求:顯示組的所有資訊,包括成員資訊,協議名稱,分配方案,訂閱資訊等。通常該請求是給管理員使用

Coordinator在rebalance的時候主要用到了前面4種請求。
4.8 liveness

consumer如何向coordinator證明自己還活著? 通過定時向coordinator傳送Heartbeat請求。如果超過了設定的超時時間,那麼coordinator就認為這個consumer已經掛了。一旦coordinator認為某個consumer掛了,那麼它就會開啟新一輪rebalance,並且在當前其他consumer的心跳response中新增“REBALANCE_IN_PROGRESS”,告訴其他consumer:不好意思各位,你們重新申請加入組吧!

4.9 Rebalance過程

終於說到consumer group執行rebalance的具體流程了。很多使用者估計對consumer內部的工作機制也很感興趣。下面就跟大家一起討論一下。當然我必須要明確表示,rebalance的前提是coordinator已經確定了。

總體而言,rebalance分為2步:Join和Sync

1 Join, 顧名思義就是加入組。這一步中,所有成員都向coordinator傳送JoinGroup請求,請求入組。一旦所有成員都發送了JoinGroup請求,coordinator會從中選擇一個consumer擔任leader的角色,並把組成員資訊以及訂閱資訊發給leader——注意leader和coordinator不是一個概念。leader負責消費分配方案的制定。

2 Sync,這一步leader開始分配消費方案,即哪個consumer負責消費哪些topic的哪些partition。一旦完成分配,leader會將這個方案封裝進SyncGroup請求中發給coordinator,非leader也會發SyncGroup請求,只是內容為空。coordinator接收到分配方案之後會把方案塞進SyncGroup的response中發給各個consumer。這樣組內的所有成員就都知道自己應該消費哪些分割槽了。

還是拿幾張圖來說明吧,首先是加入組的過程:

值得注意的是, 在coordinator收集到所有成員請求前,它會把已收到請求放入一個叫purgatory(煉獄)的地方。記得國內有篇文章以此來證明kafka開發人員都是很有文藝範的,寫得也是比較有趣,有興趣可以去搜搜。
然後是分發分配方案的過程,即SyncGroup請求:

注意!! consumer group的分割槽分配方案是在客戶端執行的!Kafka將這個權利下放給客戶端主要是因為這樣做可以有更好的靈活性。比如這種機制下我可以實現類似於Hadoop那樣的機架感知(rack-aware)分配方案,即為consumer挑選同一個機架下的分割槽資料,減少網路傳輸的開銷。Kafka預設為你提供了兩種分配策略:range和round-robin。由於這不是本文的重點,這裡就不再詳細展開了,你只需要記住你可以覆蓋consumer的引數:partition.assignment.strategy來實現自己分配策略就好了。

4.10 consumer group狀態機

和很多kafka元件一樣,group也做了個狀態機來表明組狀態的流轉。coordinator根據這個狀態機會對consumer group做不同的處理,如下圖所示(完全是根據程式碼註釋手動畫的,多見諒吧)

簡單說明下圖中的各個狀態:

  • Dead:組內已經沒有任何成員的最終狀態,組的元資料也已經被coordinator移除了。這種狀態響應各種請求都是一個response: UNKNOWN_MEMBER_ID
  • Empty:組內無成員,但是位移資訊還沒有過期。這種狀態只能響應JoinGroup請求
  • PreparingRebalance:組準備開啟新的rebalance,等待成員加入
  • AwaitingSync:正在等待leader consumer將分配方案傳給各個成員
  • Stable:rebalance完成!可以開始消費了~

至於各個狀態之間的流程條件以及action,這裡就不具體展開了。

三、rebalance場景剖析

上面詳細闡述了consumer group是如何執行rebalance的,可能依然有些雲裡霧裡。這部分對其中的三個重要的場景做詳盡的時序展開,進一步加深對於consumer group內部原理的理解。由於圖比較直觀,所有的描述都將以圖的方式給出,不做過多的文字化描述了。

1 新成員加入組(member join) 

2 組成員崩潰(member failure)

前面說過了,組成員崩潰和組成員主動離開是兩個不同的場景。因為在崩潰時成員並不會主動地告知coordinator此事,coordinator有可能需要一個完整的session.timeout週期才能檢測到這種崩潰,這必然會造成consumer的滯後。可以說離開組是主動地發起rebalance;而崩潰則是被動地發起rebalance。okay,直接上圖: 

3 組成員主動離組(member leave group)

4 提交位移(member commit offset)

總結一下,本文著重討論了一下新版本的consumer group的內部設計原理,特別是consumer group與coordinator之間的互動過程,希望對各位有所幫助。