建站
咨詢
售后
建站咨詢
新聞中心
周日早上醒來，明媚的陽光從臥室的窗戶直射進來，久違的好天氣。穿好衣服我開始籌劃今天去哪里轉(zhuǎn)轉(zhuǎn)。一周忙碌的工作幾乎沒有時間陪家人，今天該好好陪陪家人了。
10年的江城網(wǎng)站建設(shè)經(jīng)驗，針對設(shè)計、前端、開發(fā)、售后、文案、推廣等六對一服務(wù)，響應(yīng)快，48小時及時工作處理。成都全網(wǎng)營銷的優(yōu)勢是能夠根據(jù)用戶設(shè)備顯示端的尺寸不同，自動調(diào)整江城建站的顯示方式，使網(wǎng)站能夠適用不同顯示終端，在瀏覽器中調(diào)整網(wǎng)站的寬度，無論在任何一種瀏覽器上瀏覽網(wǎng)站，都能展現(xiàn)優(yōu)雅布局與設(shè)計，從而大程度地提升瀏覽體驗。成都創(chuàng)新互聯(lián)從事“江城網(wǎng)站設(shè)計”,“江城網(wǎng)站推廣”以來，每個客戶項目都認(rèn)真落實執(zhí)行。
當(dāng)我起來收拾好一切準(zhǔn)備出發(fā)的時候，我瞄了一眼手機，發(fā)現(xiàn)手機的郵箱里有一份報警郵件，報警郵件顯示線上最近10分鐘流量有異常，而且是多個渠道。有突然有一種不祥的預(yù)感：線上kafka出問題了。我讓媳婦和孩子下樓在車?yán)锏任?，我趕緊打開了電腦，查看線上系統(tǒng)。果然不出意外，kafka已經(jīng)積壓了幾千萬的數(shù)據(jù)。因為我們的業(yè)務(wù)分為實時數(shù)據(jù)和離線數(shù)據(jù)，實時數(shù)據(jù)是FileBeat負(fù)責(zé)收集日志發(fā)到Kafka，然后我們這個業(yè)務(wù)系統(tǒng)消費Kafka統(tǒng)計數(shù)據(jù)，實時數(shù)據(jù)對于當(dāng)前流量分析、預(yù)算控制、熔斷有非常重要的作用，如果實時數(shù)據(jù)異常，其它業(yè)務(wù)系統(tǒng)都會受到一定的影響。
定位到報警郵件是由于kafka消息積壓而導(dǎo)致實時數(shù)據(jù)異常觸發(fā)的，我立馬連上了我們消費Kafka的業(yè)務(wù)系統(tǒng)(data-collect)。這是一個運行時間很長了的Java服務(wù)，它的作用就是實時消費kafka數(shù)據(jù)，然后經(jīng)過一定的業(yè)務(wù)邏輯處理，將最終結(jié)果更新到mongodb中。進到服務(wù)器以后，我發(fā)現(xiàn)這個服務(wù)已經(jīng)處于假死狀態(tài)，最后一條日志顯示系統(tǒng)發(fā)生了OOM，也就是服務(wù)器內(nèi)存爆了。
關(guān)于data-collect這個Java服務(wù)的核心邏輯我在這里詳細說明一下。這個系統(tǒng)的代碼是很早的一位同事寫的，因為早期我們的數(shù)據(jù)體量還不是很大，所以，他采用了一種簡單的處理方式。先消費數(shù)據(jù)，處理完成以后放到一個Map中，然后，啟動了一個每10s執(zhí)行一次的定時任務(wù)，定時任務(wù)讀取Map數(shù)據(jù)更新到mongo中，然后清空Map(ConcurrentMap)。這樣做的優(yōu)點是將消費Kafka的操作和入庫操作分開了，可以防止因為入庫時間太長而導(dǎo)致消費速度變慢，但是，這種做法有一個致命的缺點：內(nèi)存不可控。如果定時任務(wù)因為Mongo操作時間太長而沒有及時清空Map，Map中會積累大量的數(shù)據(jù)，最終耗盡內(nèi)存，系統(tǒng)發(fā)生OOM。這時候如果系統(tǒng)自啟了，也會丟失大量的數(shù)據(jù)。
其實，這個問題我很早有意識到，但是系統(tǒng)一直運行良好，沒有出現(xiàn)任何問題，我們認(rèn)為在現(xiàn)有數(shù)據(jù)體量下它是安全的。而碰巧的是，就在前一天我們升級了Mongo的配置，mongo機器進行了一個主從切換。同時，有一些大表清理和TTL索引重建的工作還在mongo后臺運行。這就導(dǎo)致了我們操作mongo耗時的增加。進而導(dǎo)致了我們一直認(rèn)為安全的系統(tǒng)出現(xiàn)了這個問題。
回到data-collect這個系統(tǒng)的設(shè)計上?？赡苡械耐瑢W(xué)會在這里有個疑問，為什么不直接消費出來就入庫操作呢?這里我們有一個重要的處理邏輯：為了防止頻繁的更新mongo，我們會將消費出來的數(shù)據(jù)在內(nèi)存中進行一個合并處理，你可以簡單的理解為一個Map，如果key存在，我們就進行++的操作。最終操作mongo是$inc的操作，不是insert和update的操作。這也是我們需要一個ConcurrentMap的原因。也就是我們大概消費了1000萬條數(shù)據(jù)，但是最終我們處理完成以后只有10萬條數(shù)據(jù)，很多key相同的數(shù)據(jù)我們都進行了合并處理，這樣我們mongo的操作就減少了很多。
data-collect發(fā)生了OOM，我只能第一時間重啟，重啟以后，消費正常，系統(tǒng)開始有了數(shù)據(jù)。但是大概運行了幾分鐘以后，又發(fā)生了OOM。原因很簡單：kafka積壓的了大量的消息，消費很快，但是異步如mongo太耗時，所以導(dǎo)致數(shù)據(jù)全部擠壓在了這個Map內(nèi)存中?？吹竭@里，我想只能動手改造代碼了。改造的最終要達到的結(jié)果是：系統(tǒng)在不發(fā)生OOM的前提下，消費積壓在kafka中的數(shù)據(jù)，完成mongo操作。
改造的思路很簡單，就是干掉定時任務(wù)。在消費kafka消息中增加一個邏輯，每當(dāng)消費消息并且內(nèi)存進行數(shù)據(jù)合并完成以后，我們判斷Map的大小，如果Map的大小超過我們設(shè)定的限制以后，開始觸發(fā)mongo操作。之前的代碼mongo操作是單線程執(zhí)行，為了提升mongo插入操作，我們開啟20個線程并行執(zhí)行，所以我們這里需要一個帶阻塞隊列的線程池。改造后的代碼如下：
這里是SpringBoot集成Kafka的消費代碼。
這是內(nèi)存處理完成以后入mongo的操作。因為我們的topic有20個分區(qū)，所以代碼中的listenPartition0是多線程執(zhí)行的。如果沒有synchronized的同步代碼塊，那assembleyAdxTrafficVo方法就會多線程執(zhí)行，這就會導(dǎo)致數(shù)據(jù)重復(fù)插入mongo，具體大家可以體會assembleyAdxTrafficVo方法的邏輯。
而插入mongo操作的用了線程池ExecutorService，注意這里我們executorService對象的定義。
為什么要自己定義一個阻塞隊列CustomeBlockQueue?這相比很多人碰到過這個問題，如果采用默認(rèn)的阻塞隊列，例如：ArrayBlockingQueue，當(dāng)隊列長度長度超過設(shè)置的值時，ArrayBlockingQueue會拒絕新的數(shù)據(jù)進入，并且拋出異常，所以我們需要自己定義CustomeBlockQueue，并且重寫他的offer方法(BlockingQueue默認(rèn)采用offer方法將元素增加到隊列)，offer方法不會阻塞，put方法會阻塞，所以我們需要重寫offer方法，并且內(nèi)部采用put方法實現(xiàn)。關(guān)于這一點，大家可以多嘗試。ArrayBlockingQueue和LinkedBlockingQueue都有很多坑等大家去踩。
按照上述代碼處理完成上線以后，系統(tǒng)開始正常運行，kafka積壓的消息也開始慢慢降低，系統(tǒng)趨于恢復(fù)正常，而這時已經(jīng)是12點了，驚心動魄的2小時總算過去了，阿彌陀佛。
本文標(biāo)題：記一次線上Java程序故障驚心動魄兩小時
標(biāo)題URL：http://www.dlmjj.cn/article/dpdcgso.html
