摘要：同時它也支持多線程的方式，每個線程消費指定分區(qū)進(jìn)行消費。我們可以在一個消費組中創(chuàng)建多個消費實例來達(dá)到高可用高容錯的特性，不會出現(xiàn)單線程以及獨立消費者掛掉之后數(shù)據(jù)不能消費的情況。

之前寫過一篇《從源碼分析如何優(yōu)雅的使用 Kafka 生產(chǎn)者》 ，有生產(chǎn)者自然也就有消費者。

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

就我的使用經(jīng)驗來說，大部分情況都是處于數(shù)據(jù)下游的消費者角色。也用 Kafka 消費過日均過億的消息（不得不佩服 Kakfa 的設(shè)計），本文將借助我使用 Kakfa 消費數(shù)據(jù)的經(jīng)驗來聊聊如何高效的消費數(shù)據(jù)。

以之前生產(chǎn)者中的代碼為例，事先準(zhǔn)備好了一個 Topic:data-push，3個分區(qū)。

先往里邊發(fā)送 100 條消息，沒有自定義路由策略，所以消息會均勻的發(fā)往三個分區(qū)。

先來談?wù)勛詈唵蔚膯尉€程消費，如下圖所示：

由于數(shù)據(jù)散列在三個不同分區(qū)，所以單個線程需要遍歷三個分區(qū)將數(shù)據(jù)拉取下來。

單線程消費的示例代碼：

這段代碼大家在官網(wǎng)也可以找到：將數(shù)據(jù)取出放到一個內(nèi)存緩沖中最后寫入數(shù)據(jù)庫的過程。

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

通過消費日志可以看出：

取出的 100 條數(shù)據(jù)確實是分別遍歷了三個分區(qū)。

單線程消費雖然簡單，但存在以下幾個問題：

效率低下。如果分區(qū)數(shù)幾十上百個，單線程無法高效的取出數(shù)據(jù)。

可用性很低。一旦消費線程阻塞，甚至是進(jìn)程掛掉，那么整個消費程序都將出現(xiàn)問題。

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

在多線程之前不得不將消費模式分為兩種進(jìn)行探討：消費組、獨立消費者。

這兩種消費模式對應(yīng)的處理方式有著很大的不同，所以很有必要多帶帶來講。

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

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

值得注意的是：獨立消費者可以不設(shè)置 group.id 屬性。

也是發(fā)送100條消息，消費結(jié)果如下：

通過 API 可以看出：我們可以手動指定需要消費哪些分區(qū)。

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

同時它也支持多線程的方式，每個線程消費指定分區(qū)進(jìn)行消費。

為了直觀，只發(fā)送了 10 條數(shù)據(jù)。

根據(jù)消費結(jié)果可以看出：

c1 線程只取 0 分區(qū)；c2 只取 1 分區(qū)；c3 只取 2 分區(qū)的數(shù)據(jù)。

甚至我們可以將消費者多進(jìn)程部署，這樣的消費方式如下：

假設(shè) Topic:data-push 的分區(qū)數(shù)為 4 個，那我們就可以按照圖中的方式創(chuàng)建兩個進(jìn)程。

每個進(jìn)程內(nèi)有兩個線程，每個線程再去消費對應(yīng)的分區(qū)。

這樣當(dāng)我們性能不夠新增 Topic 的分區(qū)數(shù)時，消費者這邊只需要這樣水平擴(kuò)展即可，非常的靈活。

這種自定義分區(qū)消費的方式在某些場景下還是適用的，比如生產(chǎn)者每次都將某一類的數(shù)據(jù)只發(fā)往一個分區(qū)。這樣我們就可以只針對這一個分區(qū)消費。

但這種方式有一個問題：可用性不高，當(dāng)其中一個進(jìn)程掛掉之后；該進(jìn)程負(fù)責(zé)的分區(qū)數(shù)據(jù)沒法轉(zhuǎn)移給其他進(jìn)程處理。

消費組模式應(yīng)當(dāng)是使用最多的一種消費方式。

我們可以創(chuàng)建 N 個消費者實例（new KafkaConsumer()）,當(dāng)這些實例都用同一個 group.id 來創(chuàng)建時，他們就屬于同一個消費組。

在同一個消費組中的消費實例可以收到消息，但一個分區(qū)的消息只會發(fā)往一個消費實例。

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

某個 Topic 有四個分區(qū) p0 p1 p2 p3，同時創(chuàng)建了兩個消費組 groupA，groupB。

A 消費組中有兩個消費實例 C1、C2。

B 消費組中有四個消費實例 C3、C4、C5、C6。

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

通過圖中可以得知：

A 組中的 C1 消費了 P0 和 P3 分區(qū)；C2 消費 P1、P2 分區(qū)。

B 組有四個實例，所以每個實例消費一個分區(qū)；也就是消費實例和分區(qū)是一一對應(yīng)的。

需要注意的是：

這里的消費實例簡單的可以理解為 new KafkaConsumer，它和進(jìn)程沒有關(guān)系。

比如說某個 Topic 有三個分區(qū)，但是我啟動了兩個進(jìn)程來消費它。

其中每個進(jìn)程有兩個消費實例，那其實就相當(dāng)于有四個實例了。

這時可能就會問 4 個實例怎么消費 3 個分區(qū)呢？

這個 Kafka 已經(jīng)幫我做好了，它會來做消費組里的 Rebalance。

比如上面的情況，3 個分區(qū)卻有 4 個消費實例；最終肯定只有三個實例能取到消息。但至于是哪三個呢，這點 Kakfa 會自動幫我們分配好。

看個例子，還在之前的 data-push 這個 Topic，其中有三個分區(qū)。

當(dāng)其中一個進(jìn)程（其中有三個線程，每個線程對應(yīng)一個消費實例）時，消費結(jié)果如下：

里邊的 20 條數(shù)據(jù)都被這個進(jìn)程的三個實例消費掉。

這時我新啟動了一個進(jìn)程，程序和上面那個一模一樣；這樣就相當(dāng)于有兩個進(jìn)程，同時就是 6 個實例。

我再發(fā)送 10 條消息會發(fā)現(xiàn)：

進(jìn)程1 只取到了分區(qū) 1 里的兩條數(shù)據(jù)（之前是所有數(shù)據(jù)都是進(jìn)程1里的線程獲取的）。

同時進(jìn)程2則消費了剩下的 8 條消息，分別是分區(qū) 0、2 的數(shù)據(jù)（總的還是只有三個實例取到了數(shù)據(jù)，只是分別在不同的進(jìn)程里）。

當(dāng)我關(guān)掉進(jìn)程2，再發(fā)送10條數(shù)據(jù)時會發(fā)現(xiàn)所有數(shù)據(jù)又被進(jìn)程1里的三個線程消費了。

通過這些測試相信大家已經(jīng)可以看到消費組的優(yōu)勢了。

我們可以在一個消費組中創(chuàng)建多個消費實例來達(dá)到高可用、高容錯的特性，不會出現(xiàn)單線程以及獨立消費者掛掉之后數(shù)據(jù)不能消費的情況。同時基于多線程的方式也極大的提高了消費效率。

而當(dāng)新增消費實例或者是消費實例掛掉時 Kakfa 會為我們重新分配消費實例與分區(qū)的關(guān)系就被稱為消費組 Rebalance。

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

消費組中新增消費實例。

消費組中消費實例 down 掉。

訂閱的 Topic 分區(qū)數(shù)發(fā)生變化。

如果是正則訂閱 Topic 時，匹配的 Topic 數(shù)發(fā)生變化也會導(dǎo)致 Rebalance。

所以推薦使用這樣的方式消費數(shù)據(jù)，同時擴(kuò)展性也非常好。當(dāng)性能不足新增分區(qū)時只需要啟動新的消費實例加入到消費組中即可。

本次只分享了幾個不同消費數(shù)據(jù)的方式，并沒有著重研究消費參數(shù)、源碼；這些內(nèi)容感興趣的話可以在下次分享。

文中提到的部分源碼可以在這里查閱：

https://github.com/crossoverJie/JCSprout

歡迎關(guān)注公眾號一起交流：