救命！Java Web項目中的MQ消息堆積讓我抓狂
我之前參與開發(fā)了一個餐廳系統(tǒng)，該系統(tǒng)在午餐和晚餐高峰時段面臨巨大的并發(fā)需求。
為了確保系統(tǒng)順暢運行，公司要求各個部門在用餐時間輪流值班，以便及時解決在線問題。
我所在的團隊負(fù)責(zé)廚房展示系統(tǒng)，這是訂單系統(tǒng)的下游服務(wù)。
當(dāng)用戶下單后，訂單系統(tǒng)會向我們的系統(tǒng)發(fā)送一條 Kafka 消息。我們的系統(tǒng)讀取消息，處理業(yè)務(wù)邏輯，保存訂單和菜品數(shù)據(jù)，然后在菜品管理客戶端上顯示出來。
通過這種方式，廚師們可以了解每個訂單所需的菜品，一旦有菜品準(zhǔn)備好，系統(tǒng)會通知服務(wù)員上菜。
上菜后，服務(wù)員會更新菜品狀態(tài)，這樣用戶就能知道哪些菜已上，哪些菜還在準(zhǔn)備中。
這個系統(tǒng)極大地提高了從廚房到顧客的效率。
圖片
這其中的關(guān)鍵是消息中間件 Kafka。如果它出現(xiàn)問題，將直接影響廚房展示系統(tǒng)的正常運作。
在本文中，我將分享我們在處理“消息積壓問題”時的經(jīng)驗，希望能對你有所幫助。
初次遇到消息積壓問題
最初，我們的用戶量較小，系統(tǒng)上線后的一段時間里，消息隊列（MQ）的消息通信非常順暢。
隨著用戶量的增長，每個商戶每天都生成大量訂單數(shù)據(jù)，每個訂單包含多個菜品。這導(dǎo)致我們菜品管理系統(tǒng)的數(shù)據(jù)量顯著增加。
某天下午，我們收到了商戶的投訴，用戶下單后，菜品列表在平板上出現(xiàn)延遲。
廚房只有在幾分鐘后才能看到這些菜品。
我們立即開始調(diào)查原因。
菜品展示延遲問題通常與 Kafka 有關(guān)，因此我們首先檢查了 Kafka。
果不其然，存在 消息積壓。
消息積壓的常見原因有以下兩種：
- MQ 消費者服務(wù)宕機。
- MQ 生產(chǎn)者產(chǎn)生消息的速率超過 MQ 消費者消費消息的速率。
我檢查了監(jiān)控系統(tǒng)，發(fā)現(xiàn) MQ 消費者服務(wù)運行正常，沒有異常。
剩下的原因可能是 MQ 消費者的消息處理速度變慢了。
接下來，我檢查了菜品管理表，發(fā)現(xiàn)只有幾十萬條記錄。
似乎有必要優(yōu)化 MQ 消費者的處理邏輯。
我在代碼中添加了一些日志，打印出 MQ 消費者中各個關(guān)鍵節(jié)點的耗時。
發(fā)現(xiàn)兩處存在明顯的延遲：
- 在 for 循環(huán)中逐條查詢數(shù)據(jù)庫的代碼。
- 進行多條件數(shù)據(jù)查詢的代碼。
我針對性地進行了優(yōu)化：
對于 for 循環(huán)逐條查詢數(shù)據(jù)庫的代碼，我改成了使用參數(shù)集合的 批量查詢。
有時，我們需要查詢指定集合中的哪些用戶已經(jīng)存在于數(shù)據(jù)庫中。實現(xiàn)方式如下：
publicList<User>queryUser(List<User> searchList){
if(CollectionUtils.isEmpty(searchList)){
returnCollections.emptyList();
}
List<User> result = Lists.newArrayList();
searchList.forEach(user -> result.add(userMapper.getUserById(user.getId())));
return result;
}如果有 50 個用戶，這種方法需要查詢數(shù)據(jù)庫 50 次。眾所周知，每次數(shù)據(jù)庫查詢都是一次遠(yuǎn)程調(diào)用。
查詢數(shù)據(jù)庫 50 次意味著需要進行 50 次遠(yuǎn)程調(diào)用，耗時非常長。
那么，如何優(yōu)化呢？
優(yōu)化后的代碼如下：
publicList<User>queryUser(List<User> searchList){
if(CollectionUtils.isEmpty(searchList)){
returnCollections.emptyList();
}
List<Long> ids = searchList.stream().map(User::getId).collect(Collectors.toList());
return userMapper.getUserByIds(ids);
}這種方式提供了一個基于用戶 ID 集合進行批量查詢的接口，僅需一次遠(yuǎn)程調(diào)用就可以獲取所有數(shù)據(jù)。
對于多條件數(shù)據(jù)查詢，我添加了一個 組合索引，解決了這一問題。
經(jīng)過這些優(yōu)化后，MQ 消費者的消息處理速度顯著提升，成功解決了消息積壓問題。
再次遇到消息積壓問題
幾個月后，我們再次遇到了消息積壓問題。
這次問題是偶發(fā)性的，只在某些時候出現(xiàn)，大部分時間沒有問題。
積壓持續(xù)時間很短，對用戶的影響較小，商戶也沒有投訴。
我檢查了菜品管理表，此時已有幾百萬條記錄。
通過監(jiān)控和 DBA 提供的每日慢查詢郵件，我發(fā)現(xiàn)了一些異常。
我注意到有些 SQL 語句的 WHERE 條件相同，但僅參數(shù)值不同，卻使用了不同的索引。
例如，order_id=123 使用索引 a，而 order_id=124 使用索引 b。
該表存在多種查詢場景，為滿足不同的業(yè)務(wù)需求，添加了多個組合索引。
MySQL 根據(jù)以下幾種因素來選擇索引：
- 通過數(shù)據(jù)采樣估算掃描的行數(shù)。掃描的行數(shù)越多，I/O 操作和 CPU 使用率越高。
- 是否使用臨時表，臨時表會影響查詢速度。
- 是否需要排序，排序也會影響查詢速度。
綜合這些因素，MySQL 優(yōu)化器選擇其認(rèn)為最合適的索引。
MySQL 優(yōu)化器通過采樣估算掃描的行數(shù)，這涉及從數(shù)據(jù)頁中選擇一些進行統(tǒng)計估算，這種方法會帶來一定的誤差。
由于 MVCC 機制，數(shù)據(jù)頁存在多個版本，例如被刪除的數(shù)據(jù)在其他事務(wù)中仍然可見，因此索引并未真正刪除。這可能導(dǎo)致統(tǒng)計數(shù)據(jù)不準(zhǔn)確，影響優(yōu)化器的決策。
這些因素可能導(dǎo)致 MySQL 在執(zhí)行 SQL 語句時 選擇了錯誤的索引。
即便使用索引 a 更為高效，MySQL 也可能使用了索引 b。
為了解決 MySQL 選擇錯誤索引的問題，我們使用了 FORCE INDEX 關(guān)鍵字，強制 SQL 查詢使用索引 a。
經(jīng)過這一優(yōu)化，消息的輕微積壓問題也得到了解決。
第三次遭遇消息積壓
六個月后，某天晚上大約六點，幾位商家投訴菜品管理系統(tǒng)出現(xiàn)延遲。
他們反饋說下單后菜品要過幾分鐘才會顯示。
檢查監(jiān)控系統(tǒng)后，我發(fā)現(xiàn) Kafka 消息再次堆積了。
我復(fù)查了 MySQL 的索引，發(fā)現(xiàn)索引是正確的，但數(shù)據(jù)查詢依然很慢。
接著查看了菜品管理表，發(fā)現(xiàn)表中的數(shù)據(jù)量在短短六個月內(nèi)竟然增長到了 3000 萬條記錄。
通常情況下，當(dāng)單表數(shù)據(jù)過多時，查詢和寫入性能都會下降。
這次查詢變慢的原因就是因為數(shù)據(jù)量過于龐大。
為了解決這個問題，我們需要：
- 實施數(shù)據(jù)庫和表的分區(qū)
- 備份歷史數(shù)據(jù)
在當(dāng)時，實施數(shù)據(jù)庫和表分區(qū)的成本太高，而且商家數(shù)量尚不足以支持這樣的解決方案。
因此，我們決定備份歷史數(shù)據(jù)。
在與產(chǎn)品經(jīng)理和 DBA 討論后，我們決定菜品管理表僅保留最近 30 天的數(shù)據(jù)，超過 30 天的數(shù)據(jù)會被移動到一個“歷史表”中。
經(jīng)過這個優(yōu)化后，菜品管理表在 30 天內(nèi)只積累了幾百萬條數(shù)據(jù)，對性能的影響較小。
這樣一來，消息堆積問題得以解決。
第四次遇到消息堆積問題
在上述優(yōu)化之后，系統(tǒng)長時間內(nèi)運行順利，沒有出現(xiàn)消息堆積的問題。
然而，一年后的一天下午，一些商家又來投訴。
我查閱了公司郵件，發(fā)現(xiàn)有大量關(guān)于 Kafka 消息堆積的監(jiān)控告警郵件。
由于我當(dāng)時正在開會，錯過了這些告警。
這次問題的時間點比較奇怪。
通常高并發(fā)都是在午餐或晚餐的高峰時段，但這次消息堆積卻發(fā)生在“下午”。
這很不尋常。
一開始，我沒有任何線索能找到問題的原因。
于是，我詢問了訂單團隊是否在下午發(fā)布了新版本或者執(zhí)行了某些特定操作。
因為我們的菜品管理系統(tǒng)是他們的下游系統(tǒng)，直接和他們的操作相關(guān)。
一位同事提到，半小時前他們進行了一個批量更新數(shù)萬個訂單狀態(tài)的作業(yè)。
更改訂單狀態(tài)會自動發(fā)送 MQ 消息。
這導(dǎo)致他們的程序在極短時間內(nèi)產(chǎn)生了大量 MQ 消息。
我們的 MQ 消費端無法快速處理這些消息，因而出現(xiàn)了消息堆積。
我們查看了 Kafka 消息堆積情況，發(fā)現(xiàn)有數(shù)十萬條消息在排隊等待處理。
為了快速提高 MQ 消費端的處理速度，我們考慮了兩個解決方案：
- 增加分區(qū)數(shù)量。
- 使用線程池處理消息。
然而，由于消息已經(jīng)堆積在現(xiàn)有分區(qū)中，增加新的分區(qū)并不會有太大幫助。
因此，我們決定重構(gòu)代碼，使用線程池來處理消息。
為了解決堆積消息，我們將線程池的核心線程數(shù)和最大線程數(shù)增加到 50。
這些參數(shù)是可以動態(tài)配置的。
經(jīng)過這個調(diào)整后，堆積的數(shù)十萬條消息在大約 20 分鐘內(nèi)被處理完畢。
這次突發(fā)的消息堆積問題得到了妥善解決。
解決問題后，我們保留了線程池的消息消費邏輯，將核心線程數(shù)設(shè)置為 8，最大線程數(shù)設(shè)置為 10。
這樣在遇到消息堆積問題時，我們可以臨時調(diào)整線程數(shù)以快速應(yīng)對，而不會對用戶造成明顯影響。
注意：使用線程池消費 MQ 消息并不是一個通用的解決方案。它存在一些缺點，比如可能會導(dǎo)致消息順序性問題，以及服務(wù)器 CPU 使用率的飆升風(fēng)險。此外，如果在多線程環(huán)境中調(diào)用第三方接口，可能會造成第三方服務(wù)的超負(fù)荷甚至崩潰。
總結(jié)來說，MQ 消息堆積不是一個簡單的問題。
根本原因是 MQ 生產(chǎn)端的消息生產(chǎn)速率超過了消費端的消息消費速率，但具體原因可能有多種。
在實際場景中，我們需要根據(jù)不同的業(yè)務(wù)情況進行優(yōu)化。
對 MQ 隊列消息堆積的監(jiān)控和告警至關(guān)重要，能夠及時發(fā)現(xiàn)問題。
沒有完美的解決方案，只有最適合當(dāng)前業(yè)務(wù)場景的方案。
