前言

我的上家公司是做餐飲系統的，每天中午和晚上用餐高峰期，系統的并發量不容小覷。為了保險起見，公司規定各部門都要在吃飯的時間輪流值班，防止出現線上問題時能夠及時處理。

我當時在后廚顯示系統團隊，該系統屬于訂單的下游業務。

用戶點完菜下單后，訂單系統會通過發 Kafka 消息給我們系統；

系統讀取消息后，做業務邏輯處理，持久化訂單和菜品數據，然后展示到劃菜客戶端；

這樣廚師就知道哪個訂單要做哪些菜，有些菜做好了，就可以通過該系統出菜；

系統自動通知服務員上菜；

如果服務員上完菜，修改菜品上菜狀態，用戶就知道哪些菜已經上了，哪些還沒有上。

這個系統可以大大提高后廚到用戶的效率。

事實證明，這一切的關鍵是消息中間件：Kafka。如果它有問題，將會直接影響到后廚顯示系統的功能。

接下來，我跟大家一起聊聊使用 Kafka 兩年時間踩過哪些坑？

1. 順序問題

1.1 為什么要保證消息的順序？

剛開始我們系統的商戶很少，為了快速實現功能，我們沒想太多。既然是走消息中間件 Kafka 通信，訂單系統發消息時將訂單詳細數據放在消息體，我們后廚顯示系統只要訂閱 topic，就能獲取相關消息數據，然后處理自己的業務即可。

不過這套方案有個關鍵因素：要保證消息的順序。

為什么呢？

訂單有很多狀態，比如下單、支付、完成、撤銷等。不可能下單的消息都沒讀取到，就先讀取支付或撤銷的消息吧。如果真的這樣，數據不是會產生錯亂？

好吧，看來保證消息順序是有必要的。

1.2 如何保證消息順序？

我們都知道 Kafka 的 topic 是無序的，但是一個 topic 包含多個 partition，每個 partition 內部是有序的。

如此一來，思路就變得清晰了：只要保證生產者寫消息時，按照一定的規則寫到同一個 partition。不同的消費者讀不同的 partition 的消息，就能保證生產和消費者消息的順序。

我們剛開始就是這么做的，同一個商戶編號的消息寫到同一個 partition。topic 中創建了 4 個 partition，然后部署了 4 個消費者節點，構成消費者組。一個 partition 對應一個消費者節點。

從理論上說，這套方案是能夠保證消息順序的。

一切規劃得看似“天衣無縫”，我們就這樣”順利“上線了。

1.3 出現意外

該功能上線了一段時間，剛開始還是比較正常的。

但是，好景不長，很快就收到用戶投訴，說在劃菜客戶端有些訂單和菜品一直看不到，無法劃菜。

我定位到了原因，公司在那段時間網絡經常不穩定，業務接口時不時報超時，業務請求時不時會連不上數據庫。

這種情況對順序消息的打擊，可以說是毀滅性的。

為什么這么說？

假設訂單系統發了“下單”、“支付”、“完成” 三條消息。

而”下單“消息由于網絡原因我們系統處理失敗了，而后面的兩條消息的數據是無法入庫的。因為只有”下單“消息的數據才是完整的數據，其他類型的消息只會更新狀態。

加上我們當時沒有做失敗重試機制，使得這個問題被放大了。問題變成：一旦“下單”消息的數據入庫失敗，用戶就永遠看不到這個訂單和菜品了。

那么這個緊急的問題要如何解決呢？

1.4 解決過程

最開始我們的想法是：在消費者處理消息時，如果處理失敗了，立馬重試 3-5 次。

但如果有些請求要第 6 次才能成功怎么辦？

不可能一直重試呀，這種同步重試機制，會阻塞其他商戶訂單消息的讀取。

顯然，用上面的這種同步重試機制在出現異常的情況，會嚴重影響消息消費者的消費速度，降低它的吞吐量。

如此看來，我們不得不用異步重試機制了。

如果用異步重試機制，處理失敗的消息就得保存到重試表下來。

但有個新問題立馬出現：只存一條消息如何保證順序？

存一條消息的確無法保證順序，假如“下單”消息失敗了，還沒來得及異步重試。此時，“支付”消息被消費了，它肯定是不能被正常消費的。

此時，“支付”消息該一直等著，每隔一段時間判斷一次，它前面的消息都有沒有被消費？

如果真的這么做，會出現兩個問題：

“支付”消息前面只有“下單”消息，這種情況比較簡單。但如果某種類型的消息，前面有 N 多種消息，需要判斷多少次呀？這種判斷跟訂單系統的耦合性太強了，相當于要把他們系統的邏輯搬一部分到我們系統；

影響消費者的消費速度。

這時有種更簡單的方案浮出水面：消費者在處理消息時，先判斷該訂單號在重試表有沒有數據，如果有則直接把當前消息保存到重試表；如果沒有，則進行業務處理，如果出現異常，把該消息保存到重試表。

后來我們用 elastic-job 建立了失敗重試機制，如果重試了 7 次后還是失敗，則將該消息的狀態標記為失敗，發郵件通知開發人員。

終于由于網絡不穩定，導致用戶在劃菜客戶端有些訂單和菜品一直看不到的問題被解決了。現在商戶頂多偶爾延遲看到菜品，比一直看不菜品好太多。

2. 消息積壓

隨著銷售團隊的市場推廣，我們系統的商戶越來越多。隨之而來的是消息的數量越來越大，導致消費者處理不過來，經常出現消息積壓的情況。

對商戶的影響非常直觀，劃菜客戶端上的訂單和菜品可能半個小時后才能看到。一兩分鐘還能忍，半個消息的延遲，對有些暴脾氣的商戶哪里忍得了，馬上投訴過來了。我們那段時間經常接到商戶投訴說訂單和菜品有延遲。

雖說加服務器節點就能解決問題，但是按照公司為了省錢的慣例，要先做系統優化，所以我們開始了消息積壓問題解決之旅。

2.1 消息體過大

雖說 Kafka 號稱支持百萬級的 TPS，但從 producer 發送消息到 broker 需要一次網絡 IO，broker 寫數據到磁盤需要一次磁盤 IO（寫操作），consumer 從 broker 獲取消息先經過一次磁盤 IO（讀操作），再經過一次網絡 IO。

一次簡單的消息從生產到消費過程，需要經過兩次網絡 IO 和兩次磁盤 IO。如果消息體過大，勢必會增加 IO 的耗時，進而影響 Kafka 生產和消費的速度。消費者速度太慢的結果，就會出現消息積壓情況。

除了上面的問題之外，消息體過大還會浪費服務器的磁盤空間。稍不注意，可能會出現磁盤空間不足的情況。

此時，我們已經到了需要優化消息體過大問題的時候。

如何優化呢？

我們重新梳理了一下業務，沒有必要知道訂單的中間狀態，只需知道一個最終狀態就可以了。

如此甚好，我們就可以這樣設計了：

訂單系統發送的消息體只用包含 id 和狀態等關鍵信息；

后廚顯示系統消費消息后，通過 id 調用訂單系統的訂單詳情查詢接口獲取數據；

后廚顯示系統判斷數據庫中是否有該訂單的數據，如果沒有則入庫，有則更新。

果然這樣調整之后，消息積壓問題很長一段時間都沒再出現。

2.2 路由規則不合理

還真別高興的太早，有天中午又有商戶投訴說訂單和菜品有延遲。我們一查 Kafka 的 topic 竟然又出現了消息積壓。

但這次有點詭異，不是所有 partition 上的消息都有積壓，而是只有一個。

剛開始，我以為是消費那個 partition 消息的節點出了什么問題導致的。但是經過排查，沒有發現任何異常。

這就奇怪了，到底哪里有問題呢？

后來，我查日志和數據庫發現：有幾個商戶的訂單量特別大，剛好這幾個商戶被分到同一個 partition，使得該 partition 的消息量比其他 partition 要多很多。

這時我們才意識到，發消息時按商戶編號路由 partition 的規則不合理。可能會導致有些 partition 消息太多消費者處理不過來，而有些 partition 卻因為消息太少，消費者出現空閑的情況。

為了避免出現這種分配不均勻的情況，我們需要對發消息的路由規則做一下調整。

我們思考了一下，用訂單號做路由相對更均勻，不會出現單個訂單發消息次數特別多的情況。除非是遇到某個人一直加菜的情況，但是加菜是需要花錢的，所以其實同一個訂單的消息數量并不多。

調整后按訂單號路由到不同的 partition，同一個訂單號的消息，每次到發到同一個 partition。

調整后，消息積壓的問題又有很長一段時間都沒有再出現。我們的商戶數量在這段時間，增長的非常快，越來越多了。

2.3 批量操作引起的連鎖反應

在高并發的場景中，消息積壓問題可以說如影隨形，真的沒辦法從根本上解決。表面上看已經解決了，但后面不知道什么時候就會冒出一次。

比如這次。

有天下午，產品過來說：“有幾個商戶投訴過來了，他們說菜品有延遲，快查一下原因”。

這次問題出現得有點奇怪。

為什么這么說？

首先這個時間點就有點奇怪，平常出問題，不都是中午或者晚上用餐高峰期嗎？怎么這次問題出現在下午？

根據以往積累的經驗，我直接看了 Kafka 的 topic 的數據，果然上面消息有積壓。但這次每個 partition 都積壓了十幾萬的消息沒有消費，比以往加壓的消息數量增加了幾百倍。這次消息積壓得極不尋常。

我趕緊查服務監控看看消費者掛了沒，還好沒掛。又查服務日志沒有發現異常。這時我有點迷茫，碰運氣問了問訂單組下午發生了什么事情沒？他們說下午有個促銷活動，跑了一個 Job 批量更新過有些商戶的訂單信息。

這時，我一下子如夢初醒：是他們在 Job 中批量發消息導致的問題。怎么沒有通知我們呢？實在太坑了。

雖說知道問題的原因了，倒是眼前積壓的這十幾萬的消息該如何處理呢？

此時，如果直接調大 partition 數量是不行的，歷史消息已經存儲到4個固定的 partition，只有新增的消息才會到新的 partition。我們重點需要處理的是已有的 partition。

直接加服務節點也不行，因為 Kafka 允許同組的多個 partition 被一個 consumer 消費，但不允許一個 partition 被同組的多個 consumer 消費，可能會造成資源浪費。

看來只有用多線程處理了。

為了緊急解決問題，我改成了用線程池處理消息，核心線程和最大線程數都配置成了 50。

調整之后，果然，消息積壓數量不斷減少。

但此時有個更嚴重的問題出現：我收到了報警郵件，有兩個訂單系統的節點宕機了。

不久，訂單組的同事過來找我說，我們系統調用他們訂單查詢接口的并發量突增，超過了預計的好幾倍，導致有 2 個服務節點掛了。他們把查詢功能單獨整成了一個服務，部署了 6 個節點，掛了 2 個節點。再不處理，另外 4 個節點也會掛。訂單服務可以說是公司最核心的服務，它掛了公司損失會很大，情況萬分緊急。

為了解決這個問題，只能先把線程數調小。

幸好，線程數是可以通過 ZooKeeper 動態調整的。我把核心線程數調成了 8 個，核心線程數改成了 10 個。

后面，運維把訂單服務掛的 2 個節點重啟后恢復正常了。以防萬一，再多加了 2 個節點。為了確保訂單服務不會出現問題，就保持目前的消費速度，后廚顯示系統的消息積壓問題，1 小時候后也恢復正常了。

后來，我們開了一次復盤會，得出的結論是：

訂單系統的批量操作一定提前通知下游系統團隊；

下游系統團隊多線程調用訂單查詢接口一定要做壓測；

這次給訂單查詢服務敲響了警鐘。它作為公司的核心服務，應對高并發場景做的不夠好，需要做優化；

對消息積壓情況加監控。

順便說一下，對于要求嚴格保證消息順序的場景，可以將線程池改成多個隊列，每個隊列用單線程處理。

2.4 表過大

為了防止后面再次出現消息積壓問題，消費者后面就一直用多線程處理消息。

但有天中午我們還是收到很多報警郵件，提醒我們 Kafka 的 topic 消息有積壓。我們正在查原因，此時產品跑過來說：“又有商戶投訴說菜品有延遲，趕緊看看”。

這次她看起來有些不耐煩，確實優化了很多次還是出現了同樣的問題。

在外行看來：為什么同一個問題一直解決不了？

其實技術心里的苦他們是不知道的。

表面上問題的癥狀是一樣的，都是出現了菜品延遲。他們知道的是因為消息積壓導致的，但是他們不知道深層次的原因。導致消息積壓的原因其實有很多種，這也許是使用消息中間件的通病吧。

我沉默不語，只能硬著頭皮定位原因了。

后來我查日志發現消費者消費一條消息的耗時長達 2 秒。以前是 500 毫秒，現在怎么會變成 2 秒呢？

奇怪了，消費者的代碼也沒有做大的調整，為什么會出現這種情況呢？

查了一下線上菜品表，單表數據量竟然到了幾千萬，其他的劃菜表也是一樣，現在單表保存的數據太多了。

我們組梳理了一下業務，其實菜品在客戶端只展示最近 3 天的即可。

這就好辦了，我們服務端存著多余的數據，不如把表中多余的數據歸檔。于是 DBA 幫我們把數據做了歸檔，只保留最近 7 天的數據。

如此調整后，消息積壓問題被解決了，又恢復了往日的平靜。

3. 主鍵沖突

別高興得太早了，還有其他的問題。比如報警郵件經常報出數據庫異常：Duplicate entry ‘6’ for key ‘PRIMARY’，說主鍵沖突。

出現這種問題一般是由于有兩個以上相同主鍵的 SQL，同時插入數據，第一個插入成功后，第二個插入的時候會報主鍵沖突。表的主鍵是唯一的，不允許重復。

我仔細檢查了代碼，發現代碼邏輯會先根據主鍵從表中查詢訂單是否存在，如果存在則更新狀態，不存在才插入數據，沒得問題。

這種判斷在并發量不大時，是有用的。但是如果在高并發的場景下，兩個請求同一時刻都查到訂單不存在，一個請求先插入數據，另一個請求再插入數據時就會出現主鍵沖突的異常。

解決這個問題最常規的做法是：加鎖。

我剛開始也是這樣想的，加數據庫悲觀鎖肯定是不行的，太影響性能。加數據庫樂觀鎖，基于版本號判斷，一般用于更新操作，像這種插入操作基本上不會用。

剩下的只能用分布式鎖了，我們系統在用 Redis，可以加基于 Redis 的分布式鎖，鎖定訂單號。

但后面仔細思考了一下：

加分布式鎖也可能會影響消費者的消息處理速度；

消費者依賴于 Redis，如果 Redis 出現網絡超時，我們的服務就悲劇了。

所以，我也不打算用分布式鎖。

而是選擇使用 MySQL 的 INSERT INTO 。。.ON DUPLICATE KEY UPDATE 語法：

INSERTINTOtable （column_list）VALUES （value_list）ONDUPLICATEKEYUPDATEc1 = v1，c2 = v2，。。.;

它會先嘗試把數據插入表，如果主鍵沖突的話那么更新字段。

把以前的 insert 語句改造之后，就沒再出現過主鍵沖突問題。

4. 數據庫主從延遲

不久之后的某天，又收到商戶投訴說下單后，在劃菜客戶端上看得到訂單，但是看到的菜品不全，有時甚至訂單和菜品數據都看不到。

這個問題跟以往的都不一樣，根據以往的經驗先看 Kafka 的 topic 中消息有沒有積壓，但這次并沒有積壓。

再查了服務日志，發現訂單系統接口返回的數據有些為空，有些只返回了訂單數據，沒返回菜品數據。

這就非常奇怪了，我直接過去找訂單組的同事。他們仔細排查服務，沒有發現問題。這時我們不約而同的想到，會不會是數據庫出問題了，一起去找 DBA。果然 DBA發現數據庫的主庫同步數據到從庫，由于網絡原因偶爾有延遲，有時延遲有 3 秒。

如果我們的業務流程從發消息到消費消息耗時小于 3 秒，調用訂單詳情查詢接口時，可能會查不到數據，或者查到的不是最新的數據。

這個問題非常嚴重，會導致直接我們的數據錯誤。

為了解決這個問題，我們也加了重試機制。調用接口查詢數據時，如果返回數據為空，或者只返回了訂單沒有菜品，則加入重試表。

調整后，商戶投訴的問題被解決了。

5. 重復消費

Kafka消費消息時支持三種模式：

at most once 模式：最多一次。保證每一條消息 commit 成功之后，再進行消費處理。消息可能會丟失，但不會重復；

at least once 模式：至少一次。保證每一條消息處理成功之后，再進行 commit。消息不會丟失，但可能會重復；

exactly once 模式：精確傳遞一次。將 offset 作為唯一 id 與消息同時處理，并且保證處理的原子性。消息只會處理一次，不丟失也不會重復。但這種方式很難做到。

Kafka 默認的模式是 at least once，但這種模式可能會產生重復消費的問題。所以我們的業務邏輯必須做冪等設計。

而我們的業務場景保存數據時使用了 INSERT INTO 。。.ON DUPLICATE KEY UPDATE 語法，不存在時插入，存在時更新，是天然支持冪等性的。

6. 多環境消費問題

我們當時線上環境分為：pre（預發布環境）和 prod（生產環境），兩個環境共用同一個數據庫，并且共用同一個 Kafka 集群。

需要注意的是，在配置 Kafka 的 topic 的時候，要加前綴用于區分不同環境。pre環境的以 pre_ 開頭，比如 pre_order。生產環境以 prod_開頭，比如 prod_order，防止消息在不同環境中串了。

但有次運維在 pre 環境切換節點，配置 topic 的時候，錯誤地配成了 prod 的 topic。剛好那天我們有新功能上 pre 環境，結果悲劇了：prod 的有些消息被 pre 環境的 consumer 消費了。而由于消息體做了調整，導致 pre 環境的 consumer 處理消息一直失敗。

其結果是生產環境丟了部分消息。不過還好，最后生產環境消費者通過重 置offset，重新讀取了那一部分消息解決了問題，沒有造成太大損失。

后記

除了上述問題之外，我還遇到過：

Kafka 的 consumer 使用自動確認機制，導致 CPU 使用率 100%；

Kafka 集群中的一個 broker 節點掛了，重啟后又一直掛。

這兩個問題說起來有些復雜，我就不一一列舉了。非常感謝那兩年使用消息中間件 Kafka 的經歷，雖說遇到過挺多問題，踩了很多坑，走了很多彎路，但是實打實的讓我積累了很多寶貴的經驗，快速成長了。

其實 Kafka 是一個非常優秀的消息中間件，我所遇到的絕大多數問題都并非 Kafka 自身的問題（除了 CPU 使用率 100% 是它的一個 bug 導致的之外）。

編輯：jq