前言

之前寫過一篇《從原始碼分析如何優雅的使用 Kafka 生產者》 ，有生產者自然也就有消費者。

建議對 Kakfa 還比較陌生的朋友可以先看看。

就我的使用經驗來說，大部分情況都是處於資料下游的消費者角色。也用 Kafka 消費過日均過億的訊息（不得不佩服 Kakfa 的設計），本文將藉助我使用 Kakfa 消費資料的經驗來聊聊如何高效的消費資料。

單執行緒消費

以之前生產者中的程式碼為例，事先準備好了一個 Topic:data-push，3個分割槽。

先往裡邊傳送 100 條訊息，沒有自定義路由策略，所以訊息會均勻的發往三個分割槽。

先來談談最簡單的單執行緒消費，如下圖所示：

由於資料雜湊在三個不同分割槽，所以單個執行緒需要遍歷三個分割槽將資料拉取下來。

單執行緒消費的示例程式碼：

這段程式碼大家在官網也可以找到：將資料取出放到一個記憶體緩衝中最後寫入資料庫的過程。

先不討論其中的 offset 的提交方式。

通過消費日誌可以看出：

取出的 100 條資料確實是分別遍歷了三個分割槽。

單執行緒消費雖然簡單，但存在以下幾個問題：

• 效率低下。如果分割槽數幾十上百個，單執行緒無法高效的取出資料。
• 可用性很低。一旦消費執行緒阻塞，甚至是程序掛掉，那麼整個消費程式都將出現問題。

多執行緒消費

既然單執行緒有諸多問題，那是否可以用多執行緒來提高效率呢？

在多執行緒之前不得不將消費模式分為兩種進行探討：消費組、獨立消費者。

這兩種消費模式對應的處理方式有著很大的不同，所以很有必要單獨來講。

獨立消費者模式

先從獨立消費者模式談起，這種模式相對於消費組來說用的相對小眾一些。

看一個簡單示例即可知道它的用法：

值得注意的是：獨立消費者可以不設定 group.id 屬性。

也是傳送100條訊息，消費結果如下：

通過 API 可以看出：我們可以手動指定需要消費哪些分割槽。

比如 data-push Topic 有三個分割槽，我可以手動只消費其中的 1 2 分割槽，第三個可以視情況來消費。

同時它也支援多執行緒的方式，每個執行緒消費指定分割槽進行消費。

為了直觀，只發送了 10 條資料。

根據消費結果可以看出：

c1 執行緒只取 0 分割槽；c2 只取 1 分割槽；c3 只取 2 分割槽的資料。

甚至我們可以將消費者多程序部署，這樣的消費方式如下：

假設 Topic:data-push 的分割槽數為 4 個，那我們就可以按照圖中的方式建立兩個程序。

每個程序內有兩個執行緒，每個執行緒再去消費對應的分割槽。

這樣當我們效能不夠新增 Topic 的分割槽數時，消費者這邊只需要這樣水平擴充套件即可，非常的靈活。

這種自定義分割槽消費的方式在某些場景下還是適用的，比如生產者每次都將某一類的資料只發往一個分割槽。這樣我們就可以只針對這一個分割槽消費。

但這種方式有一個問題：可用性不高，當其中一個程序掛掉之後；該程序負責的分割槽資料沒法轉移給其他程序處理。

消費組模式

消費組模式應當是使用最多的一種消費方式。

我們可以建立 N 個消費者例項（new KafkaConsumer()）,當這些例項都用同一個 group.id 來建立時，他們就屬於同一個消費組。

在同一個消費組中的消費例項可以收到訊息，但一個分割槽的訊息只會發往一個消費例項。

還是藉助官方的示例圖來更好的理解它。

某個 Topic 有四個分割槽 p0 p1 p2 p3，同時建立了兩個消費組 groupA，groupB。

• A 消費組中有兩個消費例項 C1、C2。
• B 消費組中有四個消費例項 C3、C4、C5、C6。

這樣訊息是如何劃分到每個消費例項的呢？

通過圖中可以得知：

• A 組中的 C1 消費了 P0 和 P3 分割槽；C2 消費 P1、P2 分割槽。
• B 組有四個例項，所以每個例項消費一個分割槽；也就是消費例項和分割槽是一一對應的。

需要注意的是：

這裡的消費例項簡單的可以理解為 new KafkaConsumer，它和程序沒有關係。

比如說某個 Topic 有三個分割槽，但是我啟動了兩個程序來消費它。

其中每個程序有兩個消費例項，那其實就相當於有四個例項了。

這時可能就會問 4 個例項怎麼消費 3 個分割槽呢？

消費組自平衡

這個 Kafka 已經幫我做好了，它會來做消費組裡的 Rebalance。

比如上面的情況，3 個分割槽卻有 4 個消費例項；最終肯定只有三個例項能取到訊息。但至於是哪三個呢，這點 Kakfa 會自動幫我們分配好。

看個例子，還在之前的 data-push 這個 Topic，其中有三個分割槽。

當其中一個程序（其中有三個執行緒，每個執行緒對應一個消費例項）時，消費結果如下：

裡邊的 20 條資料都被這個程序的三個例項消費掉。

這時我新啟動了一個程序，程式和上面那個一模一樣；這樣就相當於有兩個程序，同時就是 6 個例項。

我再發送 10 條訊息會發現：

程序1 只取到了分割槽 1 裡的兩條資料（之前是所有資料都是程序1裡的執行緒獲取的）。

同時程序2則消費了剩下的 8 條訊息，分別是分割槽 0、2 的資料（總的還是隻有三個例項取到了資料，只是分別在不同的程序裡）。

當我關掉程序2，再發送10條資料時會發現所有資料又被程序1裡的三個執行緒消費了。

通過這些測試相信大家已經可以看到消費組的優勢了。

我們可以在一個消費組中建立多個消費例項來達到高可用、高容錯的特性，不會出現單執行緒以及獨立消費者掛掉之後資料不能消費的情況。同時基於多執行緒的方式也極大的提高了消費效率。

而當新增消費例項或者是消費例項掛掉時 Kakfa 會為我們重新分配消費例項與分割槽的關係就被稱為消費組 Rebalance。

發生這個的前提條件一般有以下幾個：

• 消費組中新增消費例項。
• 消費組中消費例項 down 掉。
• 訂閱的 Topic 分割槽數發生變化。
• 如果是正則訂閱 Topic 時，匹配的 Topic 數發生變化也會導致 Rebalance。

所以推薦使用這樣的方式消費資料，同時擴充套件性也非常好。當效能不足新增分割槽時只需要啟動新的消費例項加入到消費組中即可。

總結

本次只分享了幾個不同消費資料的方式，並沒有著重研究消費引數、原始碼；這些內容感興趣的話可以在下次分享。

文中提到的部分原始碼可以在這裡查閱：

github.com/crossoverJi…

歡迎關注公眾號一起交流：