如果問(wèn)你，如何提高kafka隊(duì)列中的消息消費(fèi)速度呢？

答案很簡(jiǎn)單，topic多分幾個(gè)分片，然后使用消費(fèi)者組（Consumer Group）去消費(fèi)topic即可。

如果加個(gè)條件，對(duì)同一個(gè)對(duì)象的操作請(qǐng)求必須要嚴(yán)格按照順序進(jìn)行處理呢？

答案也不難，topic分片之后，生產(chǎn)者定制分發(fā)策略，保證同一對(duì)象的操作請(qǐng)求都分發(fā)到同一個(gè)分片中，這樣每個(gè)消費(fèi)者就都是在按照順序消費(fèi)各自分片中的數(shù)據(jù)啦~

如果再加上一些條件：

這個(gè)消費(fèi)者消費(fèi)速度極慢、慢到需要100ms才能處理完一條消息，即使topic分100片也不滿足不了要求；

每個(gè)對(duì)象的操作請(qǐng)求數(shù)量存在嚴(yán)重傾斜的現(xiàn)象，有的分片消息數(shù)量很大，有的分片消息量很少，可能有的分片一直積壓、有的分片卻很閑；

請(qǐng)求操作很重要，需要確保每條請(qǐng)求都被可靠消費(fèi)，要保證事務(wù)最終一致性；

數(shù)十年的老系統(tǒng)，業(yè)務(wù)錯(cuò)綜復(fù)雜，項(xiàng)目方不允許涉及業(yè)務(wù)邏輯以及整體架構(gòu)的大改…

當(dāng)上述各種條件疊加到一起，要求將消費(fèi)性能明顯提升，如果是你，會(huì)如何破局呢？

前段時(shí)間，應(yīng)業(yè)務(wù)部門(mén)的要求，給他們的一個(gè)線上歷史系統(tǒng)做個(gè)并發(fā)性能提升的方案，就遇到了上述各種要求疊加在一起的棘手情況。

先簡(jiǎn)單說(shuō)下遇到的業(yè)務(wù)場(chǎng)景：

一個(gè)互動(dòng)類(lèi)的論壇的帖子評(píng)論處理場(chǎng)景，要求每個(gè)帖子的評(píng)論請(qǐng)求操作都必須要嚴(yán)格遵循一定的順序（比如可能會(huì)有評(píng)論刪除、引用評(píng)論、回復(fù)評(píng)論等操作，所以請(qǐng)求順序必須要嚴(yán)格按照順序處理），帖子評(píng)論的操作請(qǐng)求發(fā)送到kafka里面，然后評(píng)論服務(wù)消費(fèi)kafka處理各個(gè)請(qǐng)求，這個(gè)評(píng)論消費(fèi)者服務(wù)消費(fèi)太慢，需要提升下并發(fā)效率。

增加分片與消費(fèi)者數(shù)量

正式開(kāi)始著手去整改優(yōu)化。

首先是常規(guī)調(diào)整：根據(jù)kafka自身的機(jī)制，將topic進(jìn)行分片調(diào)整，拆分為N個(gè)分片，然后增設(shè)消費(fèi)者組，在消費(fèi)者組內(nèi)部署與分片數(shù)相等的消費(fèi)者服務(wù)節(jié)點(diǎn)，這樣每個(gè)消費(fèi)者可以處理一個(gè)分片，這樣整個(gè)評(píng)論的消費(fèi)性能就會(huì)提升N倍。

那么，這里為什么要強(qiáng)調(diào)消費(fèi)者組里的服務(wù)節(jié)點(diǎn)數(shù)要等于topic分片數(shù)呢？這里提一下kafka中Consumer Group中消費(fèi)者數(shù)量與topic分片數(shù)之間的相關(guān)邏輯。

看一下不同的消費(fèi)者數(shù)量與topic分片數(shù)對(duì)應(yīng)的處理消費(fèi)場(chǎng)景：

所以說(shuō)，消費(fèi)者組里面的消費(fèi)者數(shù)量并不是越多越好，而是受到了topic的分片數(shù)量的限制的：

消費(fèi)者數(shù)量太少，會(huì)導(dǎo)致一個(gè)消費(fèi)者需要消費(fèi)多個(gè)分片的數(shù)據(jù)，造成某一個(gè)消費(fèi)者消費(fèi)壓力提升；

消費(fèi)者數(shù)量太多，會(huì)導(dǎo)致有的消費(fèi)者并不會(huì)消費(fèi)任何數(shù)據(jù)，浪費(fèi)部署資源。

也是基于這一點(diǎn)，上述我們的方案中，規(guī)劃消費(fèi)者組里的消費(fèi)者數(shù)量與topic的分片數(shù)一致，這樣可以保證每個(gè)消費(fèi)者消費(fèi)1個(gè)分片，達(dá)到最大效率協(xié)調(diào)。

再補(bǔ)充個(gè)知識(shí)點(diǎn)：為什么kafka要限制每個(gè)分片最多只能有1個(gè)消費(fèi)者組里的消費(fèi)者在處理呢？

因?yàn)橄M(fèi)者拉取消息需要提供offset, limit。如果offset放在broker端，那么一定會(huì)產(chǎn)生額外的通信開(kāi)銷(xiāo)；如果offset放在Consumer端，如果在一個(gè)組有多個(gè)消費(fèi)者，就需要有一個(gè)協(xié)調(diào)者，集中式的管理，解決鎖沖突，如果不解決沖突，那么勢(shì)必會(huì)產(chǎn)生重復(fù)消費(fèi)、無(wú)用的消費(fèi)，從而導(dǎo)致資源浪費(fèi)。所以說(shuō)，從性能與復(fù)雜度的取舍上，Kafka采用了相對(duì)簡(jiǎn)單的一種解決策略。

保證分片內(nèi)寫(xiě)入順序

通過(guò)上一章的方式，增加了topic分區(qū)數(shù)以及消費(fèi)者組中消費(fèi)者數(shù)量，對(duì)kafka中消息并行消費(fèi)的效率是提升了，但是問(wèn)題又來(lái)了：順序問(wèn)題！

前面說(shuō)過(guò)，由于業(yè)務(wù)明確要求確保順序消費(fèi)，而kafka只是保證分片內(nèi)的消費(fèi)順序是固定的，但是不同分片之間的消費(fèi)順序是無(wú)法保證的。

對(duì)業(yè)務(wù)進(jìn)行分析發(fā)現(xiàn)，業(yè)務(wù)要求的順序處理，其實(shí)是有條件的順序處理。即對(duì)于同一個(gè)帖子的所有評(píng)論相關(guān)的操作必須要同步處理，對(duì)于不同帖子的評(píng)論相關(guān)操作并沒(méi)有順序的要求。那么問(wèn)題就簡(jiǎn)單了，只要保證同一個(gè)帖子的所有評(píng)論相關(guān)操作請(qǐng)求都被分發(fā)到同一個(gè)topic分區(qū)內(nèi)即可！

生產(chǎn)者寫(xiě)入消息到kafka的topic時(shí)，kafka將依據(jù)不同的策略將數(shù)據(jù)分配到不同的分區(qū)中：

• 輪詢分區(qū)策略
• 隨機(jī)分區(qū)策略
• 按key分區(qū)分配策略
• 自定義分區(qū)策略

這里采用自定義分區(qū)策略，因?yàn)槊總€(gè)評(píng)論操作請(qǐng)求中都攜帶有一個(gè)原始帖子ID字段，所以分發(fā)策略也很簡(jiǎn)單，直接帖子ID % 分片數(shù)將消息進(jìn)行分發(fā)，這樣同一個(gè)帖子ID的評(píng)論操作就都可以到同一個(gè)分片中，這樣順序的問(wèn)題就解決了。

所以，對(duì)上一環(huán)節(jié)給定的初步方案進(jìn)行優(yōu)化，補(bǔ)充下生產(chǎn)者端的定制化分發(fā)策略的要求，保證同一個(gè)帖子的評(píng)論操作都會(huì)到同一個(gè)Topic分片中：

方案設(shè)計(jì)到這里，似乎已經(jīng)是解決了并發(fā)消費(fèi)的問(wèn)題了。但是后來(lái)實(shí)際壓測(cè)之后，結(jié)果令人大跌眼鏡。

單消費(fèi)者速度提升

按照前面給出的方案，部署了DEMO環(huán)境進(jìn)行壓測(cè)（拆分成4個(gè)分片，部署4個(gè)消費(fèi)者），最終發(fā)現(xiàn)集群消費(fèi)速度的確是翻了4倍、但是整體并發(fā)量依舊是低的可憐，4臺(tái)機(jī)器最終消費(fèi)并發(fā)量甚至不到100??？

心靈受到暴擊之后，去分析下單個(gè)消費(fèi)者節(jié)點(diǎn)的運(yùn)行情況，發(fā)現(xiàn)壓測(cè)過(guò)程中整個(gè)機(jī)器CPU、IO、MEM、線程數(shù)都非常低、毫無(wú)任何波動(dòng)。問(wèn)業(yè)務(wù)方要了代碼權(quán)限，下載了代碼并走讀了一遍Consumer服務(wù)的代碼邏輯才發(fā)現(xiàn)其中玄機(jī)。

其實(shí)該業(yè)務(wù)整體交互邏輯其實(shí)很簡(jiǎn)單，從kafka獲取一個(gè)消息，然后進(jìn)行消費(fèi)。但是這個(gè)消費(fèi)邏輯，是需要按順序調(diào)用10余個(gè)周邊系統(tǒng)的HTTP接口！這也難怪CPU、內(nèi)存、IO都非常低了，整個(gè)進(jìn)程中只有一個(gè)線程在處理業(yè)務(wù)、而這個(gè)線程大部分時(shí)間都是處于IO等待狀態(tài)。

所以要想提升整體集群的消費(fèi)能力，要么無(wú)限擴(kuò)機(jī)器、要么就提升單節(jié)點(diǎn)的消費(fèi)能力 —— 顯然前者是不可能的，只能選擇后者。而對(duì)于單線程、多IO操作的場(chǎng)景，提升并發(fā)性能，首先想到的就是改為多線程并發(fā)處理。但是多線程并發(fā)的時(shí)候，又會(huì)涉及到如何保證順序消費(fèi)的問(wèn)題。

對(duì)前面的方案進(jìn)行優(yōu)化，給出如下方案：

在前面方案的基礎(chǔ)上，主要是對(duì)消費(fèi)者端的實(shí)現(xiàn)邏輯進(jìn)行了調(diào)整：

在消費(fèi)者內(nèi)部，區(qū)分Consumer Thread和Work Thread，Consumer Thread負(fù)責(zé)從kafka拉取消息，而Work Thread負(fù)責(zé)真正的消費(fèi)邏輯處理。

單機(jī)內(nèi)存中維護(hù)若干個(gè)隊(duì)列，每個(gè)隊(duì)列對(duì)應(yīng)一個(gè)Work Thread，負(fù)責(zé)消費(fèi)該隊(duì)列中的數(shù)據(jù)；

Consumer Thread基于親緣性分發(fā)策略對(duì)消息進(jìn)行二次分發(fā)，保證相同帖子ID的請(qǐng)求分發(fā)到不同的內(nèi)部隊(duì)列中。

再進(jìn)行壓測(cè)，設(shè)置單個(gè)消費(fèi)者服務(wù)Work Thread數(shù)量為100，集群內(nèi)4個(gè)消費(fèi)者服務(wù)，整體消費(fèi)速度達(dá)到了7000。單節(jié)點(diǎn)的消費(fèi)性能從原來(lái)的20提升到1700，提升了近80倍！

如何保證消息不丟失

經(jīng)過(guò)將單機(jī)的消費(fèi)模式改為多線程的方式，目前并發(fā)消費(fèi)性能的問(wèn)題是解決了，可是可靠性的問(wèn)題又出現(xiàn)了。

原先的時(shí)候，消費(fèi)者從kafka拉取一條消息，然后消費(fèi)完成后，給kafka一個(gè)ack應(yīng)答，然后去拉取下一條消息，這樣即使消費(fèi)者中途宕機(jī)了，kafka依舊可以將消息分發(fā)給下一個(gè)可用的消費(fèi)者去處理，可以保證請(qǐng)求消息不會(huì)丟失掉。

而前面的方案，消費(fèi)者服務(wù)從kafka拉取到消息之后，并沒(méi)有等待處理完成，就繼續(xù)從kafka拉取消息然后緩存在本機(jī)內(nèi)存中等待work thread慢慢消費(fèi)，這個(gè)時(shí)候，如果機(jī)器宕機(jī)，所有緩存的消息將全部丟失！

為了解決上述問(wèn)題，考慮將kafka應(yīng)答機(jī)制改為手動(dòng)提交ack。但是由于多個(gè)線程之間亂序的處理kafka上的數(shù)據(jù)，各個(gè)線程已經(jīng)處理的offset值是不一樣的。如下示意圖：

為了保證消息可靠不丟失，采用如下策略：定期手動(dòng)提交當(dāng)前的offset信息，提交的offset值，選擇當(dāng)前節(jié)點(diǎn)已處理的最小offset值（對(duì)于上面示意圖，即提交1002這個(gè)offset值），可以通過(guò)在內(nèi)存中緩存下處理的offset列表的方式實(shí)現(xiàn)，如下如實(shí)現(xiàn)策略：

正常情況下，提交的offset值不會(huì)有什么作用或影響，但是一旦出現(xiàn)異常情況，導(dǎo)致當(dāng)前節(jié)點(diǎn)進(jìn)程不可用，kafka重平衡將當(dāng)前分片分給另一個(gè)消費(fèi)者進(jìn)行消費(fèi)的時(shí)候，另一個(gè)消費(fèi)者會(huì)從最后一次提交的offset位置開(kāi)始繼續(xù)往后消費(fèi)。這樣便解決了數(shù)據(jù)丟失的問(wèn)題，保證了數(shù)據(jù)可靠。

但是，另一個(gè)問(wèn)題又出現(xiàn)了：重復(fù)消費(fèi)。好在，雖然這個(gè)業(yè)務(wù)系統(tǒng)是十多年前構(gòu)建的，但是至少分布式消費(fèi)者該有的一個(gè)關(guān)鍵特性還是具備的，那就是冪等，所以這個(gè)問(wèn)題就不用考慮了。

數(shù)據(jù)積壓不可控場(chǎng)景兜底

到這里，總該一切都沒(méi)問(wèn)題了吧？

是，也不是。正常情況下是沒(méi)問(wèn)題了，但是作為一個(gè)"核心"系統(tǒng)，極端的異常情況的保命策略還需要考慮下。

舉個(gè)例子，如果突然有一條帖子爆火，這條帖子的評(píng)論量遠(yuǎn)超其余帖子的評(píng)論量，甚至遠(yuǎn)超整個(gè)系統(tǒng)的額定最大負(fù)載請(qǐng)求量，這樣會(huì)出現(xiàn)個(gè)問(wèn)題：

• kafka某一個(gè)分片數(shù)據(jù)量積壓嚴(yán)重，其余分片很空閑
• 該條火爆的帖子的相關(guān)評(píng)論請(qǐng)求，阻塞了與該帖子分配到同一分區(qū)的其余帖子的評(píng)論處理。

這個(gè)原計(jì)劃做一個(gè)動(dòng)態(tài)伸縮的分片分發(fā)策略，但考慮到此場(chǎng)景過(guò)于極端，當(dāng)前系統(tǒng)實(shí)施起來(lái)性價(jià)比不高，所以本著適當(dāng)設(shè)計(jì)的原則，放棄了原先方案，改為了簡(jiǎn)單的手動(dòng)處理 + 補(bǔ)償服務(wù)方式，如下：

一旦出現(xiàn)未預(yù)料到的異常，導(dǎo)致系統(tǒng)積壓已經(jīng)超過(guò)正常的處理范圍了，且已經(jīng)遠(yuǎn)超系統(tǒng)可以正?；謴?fù)的限度，為了保證現(xiàn)有業(yè)務(wù)盡快的恢復(fù)正常，可以先跳過(guò)積壓的請(qǐng)求，先保證新過(guò)來(lái)的請(qǐng)求正常被處理，然后啟動(dòng)補(bǔ)償進(jìn)程，慢慢消費(fèi)之前積壓的消息。

有一說(shuō)一：

這個(gè)地方是整個(gè)方案里面我自己不太滿意的一個(gè)實(shí)現(xiàn)，屬于遷就現(xiàn)實(shí)的一種妥協(xié)方案，寫(xiě)這篇文檔的時(shí)候，自己還是打算近期將這部分按照一個(gè)更優(yōu)的方案進(jìn)行實(shí)現(xiàn)。如果您也有興趣了解或者有更好的建議思路，歡迎聯(lián)系我，我們一起掰扯下。

總結(jié)梳理

至此呢，為了解決kafka消費(fèi)者消費(fèi)能力太慢場(chǎng)景的集群并發(fā)性能提升方案就全部設(shè)計(jì)完成了，業(yè)務(wù)要求的各種要求約束也都可以滿足了，最終實(shí)現(xiàn)了在業(yè)務(wù)邏輯沒(méi)有變的情況下，整體集群的性能提升了上百倍。整體的改動(dòng)內(nèi)容如下：