記一次多線程使用不當(dāng)，導(dǎo)致項目OOM

作者：樹洞君

鏈接：juejin.cn/post/7064376361334358046

多線程是工作中經(jīng)常要使用的，也是面試必考的，

多線程相關(guān)文章推薦閱讀：

【Java并發(fā)編程】從源碼分析幾道必問線程池的面試題？

【Java并發(fā)編程】從源碼分析幾道必問線程池的面試題？

??還在使用Future輪詢獲取結(jié)果嗎？CompletionService快來了解下。

事故描述

從6點32分開始少量用戶訪問app時會出現(xiàn)首頁訪問異常，到7點20分首頁服務(wù)大規(guī)模不可用，7點36分問題解決。

整體經(jīng)過

• 6:58 ?發(fā)現(xiàn)報警，同時發(fā)現(xiàn)群里反饋首頁出現(xiàn)網(wǎng)絡(luò)繁忙，考慮到前幾日晚上門店列表服務(wù)上線發(fā)布過，所以考慮回滾代碼緊急處理問題。
• 7:07 ?開始先后聯(lián)系XXX查看解決問題。
• 7:36 ?代碼回滾完，服務(wù)恢復(fù)正常

事故根本原因-事故代碼模擬

public?static?void?test()?throws?InterruptedException,?ExecutionException?{
????Executor?executor?=?Executors.newFixedThreadPool(3);
????CompletionService?service?=?new?ExecutorCompletionService<>(executor);
????????service.submit(new?Callable()?{
????????????@Override
????????????public?String?call()?throws?Exception?{
????????????????return?"HelloWorld--"?+?Thread.currentThread().getName();
????????????}
????????});
}

先拋出問題，我們后面會詳細闡述。問題的根源就在于ExecutorCompletionService結(jié)果沒 調(diào)用take，poll方法。正確的寫法如下所示：

public?static?void?test()?throws?InterruptedException,?ExecutionException?{
????Executor?executor?=?Executors.newFixedThreadPool(3);
????CompletionService?service?=?new?ExecutorCompletionService<>(executor);
????service.submit(new?Callable()?{
????????@Override
????????public?String?call()?throws?Exception?{
????????????return?"HelloWorld--"?+?Thread.currentThread().getName();
????????}
????});
????service.take().get();
}

一行代碼引發(fā)的血案，而且不容易被發(fā)現(xiàn)，因為oom是一個內(nèi)存緩慢增長的過程，稍微粗心大意就會忽略，如果是這個代碼塊的調(diào)用量少的話，很可能幾天甚至幾個月后暴雷。

接下來我們來探討問題的根源

為了更好的理解ExecutorCompletionService的 “套路” 我們用 ExecutorService來作為對比，可以讓我們更好的清楚，什么場景下用ExecutorCompletionService。先看ExecutorService代碼（建議down下來跑一跑 以下代碼建議吃飯的時候不要去看，味道略重！不過便于理解orz；）

public?static?void?test1()?throws?Exception{
????ExecutorService?executorService?=?Executors.newCachedThreadPool();
????ArrayList>?futureArrayList?=?new?ArrayList<>();
????System.out.println("公司讓你通知大家聚餐?你開車去接人");
????Future?future10?=?executorService.submit(()?->?{
????????System.out.println("總裁：我在家上大號?我最近拉肚子比較慢?要蹲1個小時才能出來?你等會來接我吧");
????????TimeUnit.SECONDS.sleep(10);
????????System.out.println("總裁：1小時了?我上完大號了。你來接吧");
????????return?"總裁上完大號了";

????});
????futureArrayList.add(future10);
????Future?future3?=?executorService.submit(()?->?{
????????System.out.println("研發(fā)：我在家上大號?我比較快?要蹲3分鐘就可以出來?你等會來接我吧");
????????TimeUnit.SECONDS.sleep(3);
????????System.out.println("研發(fā)：3分鐘?我上完大號了。你來接吧");
????????return?"研發(fā)上完大號了";
????});
????futureArrayList.add(future3);
????Future?future6?=?executorService.submit(()?->?{
????????System.out.println("中層管理：我在家上大號??要蹲10分鐘就可以出來?你等會來接我吧");
????????TimeUnit.SECONDS.sleep(6);
????????System.out.println("中層管理：10分鐘?我上完大號了。你來接吧");
????????return?"中層管理上完大號了";
????});
????futureArrayList.add(future6);
????TimeUnit.SECONDS.sleep(1);
????System.out.println("都通知完了,等著接吧。");
????try?{
????????for?(Future?future?:?futureArrayList)?{
????????????String?returnStr?=?future.get();
????????????System.out.println(returnStr?+?"，你去接他");
????????}
????????Thread.currentThread().join();
????}?catch?(Exception?e)?{
????????e.printStackTrace();
????}
}

三個任務(wù)，每個任務(wù)執(zhí)行時間分別是 10s、3s、6s 。通過JDK線程池的 submit 提交這三個 Callable類型的任務(wù)。

step1 主線程把三個任務(wù)提交到線程池里面去，把對應(yīng)返回的 Future 放到 List 里面存起來，然后執(zhí)行“都通知完了,等著接吧。”這行輸出語句。step2在循環(huán)里面執(zhí)行 future.get() 操作，阻塞等待最后結(jié)果如下：

先通知到總裁，也是先接總裁 足足等了1個小時，接到總裁后再去接研發(fā)和中層管理，盡管他們早就完事兒了，也得等總裁拉完~~

耗時最久的-10s異步任務(wù)最先進入list執(zhí)行，所以在循環(huán)過程中獲取這個10s的任務(wù)結(jié)果的時候，get操作會一直阻塞，直到10s異步任務(wù)執(zhí)行完畢。即使 3s、5s的任務(wù)早就執(zhí)行完了，也得阻塞等待10s任務(wù)執(zhí)行完。看到這里 尤其是做網(wǎng)關(guān)業(yè)務(wù)的同學(xué)可能會產(chǎn)生共鳴，一般來說網(wǎng)關(guān)RPC會調(diào)用下游N多個接口，如下圖

如果都按照ExecutorService這種方式，并且恰巧前幾個任務(wù)調(diào)用的接口耗時比較久，同時阻塞等待，那就比較悲催了。所以ExecutorCompletionService應(yīng)景而出。它作為任務(wù)線程的合理管控者，“任務(wù)規(guī)劃師”的稱號名副其實。

相同場景 ExecutorCompletionService代碼

public?static?void?test2()?throws?Exception?{
????ExecutorService?executorService?=?Executors.newCachedThreadPool();
????ExecutorCompletionService?completionService?=?new?ExecutorCompletionService<>(executorService);
????System.out.println("公司讓你通知大家聚餐?你開車去接人");
????completionService.submit(()?->?{
????????System.out.println("總裁：我在家上大號?我最近拉肚子比較慢?要蹲1個小時才能出來?你等會來接我吧");
????????TimeUnit.SECONDS.sleep(10);
????????System.out.println("總裁：1小時了?我上完大號了。你來接吧");
????????return?"總裁上完大號了";
????});
????completionService.submit(()?->?{
????????System.out.println("研發(fā)：我在家上大號?我比較快?要蹲3分鐘就可以出來?你等會來接我吧");
????????TimeUnit.SECONDS.sleep(3);
????????System.out.println("研發(fā)：3分鐘?我上完大號了。你來接吧");
????????return?"研發(fā)上完大號了";
????});
????completionService.submit(()?->?{
????????System.out.println("中層管理：我在家上大號??要蹲10分鐘就可以出來?你等會來接我吧");
????????TimeUnit.SECONDS.sleep(6);
????????System.out.println("中層管理：10分鐘?我上完大號了。你來接吧");
????????return?"中層管理上完大號了";
????});
????TimeUnit.SECONDS.sleep(1);
????System.out.println("都通知完了,等著接吧。");
????//提交了3個異步任務(wù)）
????for?(int?i?=?0;?i?3;?i++)?{
????????String?returnStr?=?completionService.take().get();
????????System.out.println(returnStr?+?"，你去接他");
????}
????Thread.currentThread().join();
}

跑完結(jié)果如下：

這次就相對高效了一些，雖然先通知的總裁，但是根據(jù)大家上大號的速度，誰先拉完先去接誰，不用等待上大號最久的總裁了（現(xiàn)實生活里 建議采用第一種 不等總裁的后果 emmm 哈哈哈）。

放在一起對比下輸出結(jié)果：兩段代碼的差異非常小 獲取結(jié)果的時候ExecutorCompletionService 使用了

completionService.take().get();

為毛要用take() 然后再get()呢？？？？我們看看源碼

CompletionService接口 以及接口的實現(xiàn)類

1、ExecutorCompletionService是CompletionService接口的實現(xiàn)類

接著跟一下ExecutorCompletionService的構(gòu)造方法，可以看到入?yún)⑿枰獋饕粋€線程池對象，默認(rèn)使用的隊列是 LinkedBlockingQueue，不過還有另外一個構(gòu)造方法可以指定隊列類型，如下兩張圖，兩個構(gòu)造方法。

默認(rèn)LinkedBlockingQueue的構(gòu)造方法

可選隊列類型的構(gòu)造方法

submit任務(wù)提交的兩種方式，都是有返回值的，我們例子中用到的就是第一種Callable類型的方法。

對比ExecutorService 和 ExecutorCompletionService submit方法 可以看出區(qū)別

（1）ExecutorService（2）ExecutorCompletionService

差異就在 QueueingFuture，這個到底作用是啥，我們繼續(xù)跟進去看

• QueueingFuture 繼承自 FutureTask，而且紅線部分標(biāo)注的位置，重寫了done()方法。

• 把 task 放到 completionQueue 隊列里面，當(dāng)任務(wù)執(zhí)行完成后，task就會被放到隊列里面去了。

• 此時此刻completionQueue隊列里面的 task 都是已經(jīng) done()完成了的 task，而這個 task 就是我們拿到的一個個的future結(jié)果。

• 如果調(diào)用 completionQueue 的 task 方法，會阻塞等待任務(wù)。等到的一定是完成了的 future，我們調(diào)用 .get()方法 就能立馬獲得結(jié)果。

• 看到這里 相信大家伙都應(yīng)該多少明白點了

• 我們在使用ExecutorService submit提交任務(wù)后需要關(guān)注每個任務(wù)返回的future，然而CompletionService 對這些 future 進行了追蹤，并且重寫了done方法，讓你等的completionQueue 隊列里面 一定是完成了的task。

• 作為網(wǎng)關(guān)RPC層，我們不用因為某一個接口的響應(yīng)慢拖累所有的請求，可以在處理最快響應(yīng)的業(yè)務(wù)場景里使用CompletionService。

but 注意、注意、注意 也是本次事故的核心當(dāng)只有調(diào)用了ExecutorCompletionService下面的3個方法的任意一個時，阻塞隊列中的task執(zhí)行結(jié)果才會從隊列中移除掉,釋放堆內(nèi)存，由于該業(yè)務(wù)不需要使用任務(wù)的返回值，則沒進行調(diào)用take，poll方法。從而導(dǎo)致沒有釋放堆內(nèi)存，堆內(nèi)存會隨著調(diào)用量的增加一直增長。所以，業(yè)務(wù)場景中不需要使用任務(wù)返回值的 別沒事兒使用CompletionService，假如使用了，記得一定要從阻塞隊列中 移除掉task執(zhí)行結(jié)果，避免OOM！知道事故的原因，我們來總結(jié)下方法論，畢竟孔子他老人家說過：自省吾身，常思己過，善修其身！上線前：

• 嚴(yán)格的代碼review習(xí)慣，一定要交給back人去看，畢竟自己寫的代碼自己是看不出問題的，相信每個程序猿都有這個自信（這個后續(xù)事故里可能會反復(fù)提到！很重要）
• 上線記錄-備注好上一個可回滾的包版本（給自己留一個后路）
• 上線前確認(rèn)回滾后，業(yè)務(wù)是否可降級，如果不可降級，一定要嚴(yán)格拉長這次上線的監(jiān)控周期 上線后：
• 持續(xù)關(guān)注內(nèi)存增長情況（這部分極容易被忽略，大家對內(nèi)存的重視度不如cpu使用率）
• 持續(xù)關(guān)注cpu使用率增長情況
• gc情況、線程數(shù)是否增長、是否有頻繁的fullgc等 關(guān)注服務(wù)性能報警，tp99、999 、max是否出現(xiàn)明顯的增高