一次因HashSet引起的并發(fā)問題詳解

更新時間：2018年11月08日 10:04:13 作者：crossoverjie

這篇文章主要給大家介紹了一次因HashSet引起的并發(fā)問題的解決方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

為啥要用HahSet?

假如我們現(xiàn)在想要在一大堆數(shù)據(jù)中查找X數(shù)據(jù)。LinkedList的數(shù)據(jù)結(jié)構(gòu)就不說了，查找效率低的可怕。ArrayList哪，如果我們不知道X的位置序號，還是一樣要全部遍歷一次直到查到結(jié)果，效率一樣可怕。HashSet天生就是為了提高查找效率的。

背景

上午剛到公司，準備開始一天的摸魚之旅時突然收到了一封監(jiān)控中心的郵件。

心中暗道不好，因為監(jiān)控系統(tǒng)從來不會告訴我應用完美無 bug，其實系統(tǒng)挺猥瑣。

打開郵件一看，果然告知我有一個應用的線程池隊列達到閾值觸發(fā)了報警。

由于這個應用出問題非常影響用戶體驗；于是立馬讓運維保留現(xiàn)場 dump 線程和內(nèi)存同時重啟應用，還好重啟之后恢復正常。于是開始著手排查問題。

分析

首先了解下這個應用大概是做什么的。

簡單來說就是從 MQ 中取出數(shù)據(jù)然后丟到后面的業(yè)務線程池中做具體的業(yè)務處理。

而報警的隊列正好就是這個線程池的隊列。

跟蹤代碼發(fā)現(xiàn)構(gòu)建線程池的方式如下：

ThreadPoolExecutor executor = new ThreadPoolExecutor(coreSize, maxSize,
  0L, TimeUnit.MILLISECONDS,
  new LinkedBlockingQueue<Runnable>());;
  put(poolName,executor);

采用的是默認的 LinkedBlockingQueue 并沒有指定大小（這也是個坑），于是這個隊列的默認大小為 Integer.MAX_VALUE。

由于應用已經(jīng)重啟，只能從僅存的線程快照和內(nèi)存快照進行分析。

內(nèi)存分析

先利用 MAT 分析了內(nèi)存，的到了如下報告。

其中有兩個比較大的對象，一個就是之前線程池存放任務的 LinkedBlockingQueue，還有一個則是 HashSet。

當然其中隊列占用了大量的內(nèi)存，所以優(yōu)先查看，HashSet 一會兒再看。

由于隊列的大小給的夠大，所以結(jié)合目前的情況來看應當是線程池里的任務處理較慢，導致隊列的任務越堆越多，至少這是目前可以得出的結(jié)論。

線程分析

再來看看線程的分析，這里利用fastthread.io 這個網(wǎng)站進行線程分析。

因為從表現(xiàn)來看線程池里的任務遲遲沒有執(zhí)行完畢，所以主要看看它們在干嘛。

正好他們都處于 RUNNABLE 狀態(tài)，同時堆棧如下：

發(fā)現(xiàn)正好就是在處理上文提到的 HashSet，看這個堆棧是在查詢 key 是否存在。通過查看 312 行的業(yè)務代碼確實也是如此。

這里的線程名字也是個坑，讓我找了好久。

定位

分析了內(nèi)存和線程的堆棧之后其實已經(jīng)大概猜出一些問題了。

這里其實有一個前提忘記講到：

這個告警是凌晨三點發(fā)出的郵件，但并沒有電話提醒之類的，所以大家都不知道。

到了早上上班時才發(fā)現(xiàn)并立即 dump 了上面的證據(jù)。

所有有一個很重要的事實：這幾個業(yè)務線程在查詢 HashSet 的時候運行了 6 7 個小時都沒有返回。

通過之前的監(jiān)控曲線圖也可以看出：

操作系統(tǒng)在之前一直處于高負載中，直到我們早上看到報警重啟之后才降低。

同時發(fā)現(xiàn)這個應用生產(chǎn)上運行的是 JDK1.7 ，所以我初步認為應該是在查詢 key 的時候進入了 HashMap 的環(huán)形鏈表導致 CPU 高負載同時也進入了死循環(huán)。

為了驗證這個問題再次 review 了代碼。

整理之后的偽代碼如下：

//線程池
private ExecutorService executor;
private Set<String> set = new hashSet();
private void execute(){
 
 while(true){
 //從 MQ 中獲取數(shù)據(jù)
 String key = subMQ();
 executor.excute(new Worker(key)) ;
 }
}
public class Worker extends Thread{
 private String key ;
 public Worker(String key){
 this.key = key;
 }
 @Override
 private void run(){
 if(!set.contains(key)){
 //數(shù)據(jù)庫查詢
 if(queryDB(key)){
 set.add(key);
 return;
 }
 }
 //達到某種條件時清空 set
 if(flag){
 set = null ;
 }
 } 
}

大致的流程如下：