本文轉載自微信公眾號「 三太子敖丙」，轉載本文請聯系 三太子敖丙公眾號。

[[329971]]

 背景

敖丙之前在工作中遇到一個問題，我定義了一個線程池來執行任務，但是程序執行結束后任務沒有全部執行完，當時心態就差點崩了。

業務場景是這樣的：由于統計業務需要，訂單信息需要從主庫中經過統計業務代碼寫入統計庫(中間需要邏輯處理所以不能走binlog)。

由于代碼質量及歷史原因，目前的重新統計接口是單線程的，粗略算了算一共有100萬條訂單信息，每100條的處理大約是10秒，所以理論上處理完全部信息需要28個小時，這還不算因為 mysql 中 limit 分頁導致的后期查詢時間以及可能出現的內存溢出導致中止統計的情況。

基于上述的原因，以及最重要的一點：統計業務是根據訂單所屬的中心進行的，各個中心同時統計不會導致臟數據。

所以，我計劃使用線程池，為每一個中心分配一條線程去執行統計業務。

業務實現

// 線程工廠，用于為線程池中的每條線程命名 
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder(）.setNameFormat("stats-pool-%d").build(); 
 
// 創建線程池，使用有界阻塞隊列防止內存溢出 
ExecutorService statsThreadPool = new ThreadPoolExecutor(5, 10, 
                0L, TimeUnit.MILLISECONDS, 
                new LinkedBlockingQueue<>(100), namedThreadFactory); 
// 遍歷所有中心，為每一個centerId提交一條任務到線程池 
statsThreadPool.submit(new StatsJob(centerId)); 

在創建完線程池后，為每一個 centerId 提交一條任務到線程池，在我的預想中，由于線程池的核心線程數為5，最多5個中心同時進行統計業務，將大大縮短100萬條數據的總統計時間，于是萬分興奮的我開始執行重新統計業務了。

問題

在跑了很久之后，當我查看統計進度時，我發現了一個十分詭異的問題(如下圖)。

藍框標出的這條線程是 WAIT 狀態，表明這條線程是空閑狀態，但是從日志中我看到這條線程并沒有完成它的任務，因為這個中心的數據有10萬條，但是日志顯示它只跑到了一半，之后就再無關于此中心的日志了。

這是什么原因?

我當場就想到了三歪，肯定是三歪今天早上上班左腳先邁進公司的，導致代碼水土不服，一定是這樣，我去找他去。

調試及原因

咳咳三歪是開玩笑的，我們還是需要找到真實原因。

可以想到的是，這條線程因為某些原因被阻塞了，并且沒有繼續進行下去，但是日志又沒有任何異常信息…

可能有經驗的工程師已經知道了原因…

由于個人水平的線程，暫時沒有找到原因的我只能放棄使用線程池，乖乖用單線程跑…

幸運的是，單線程跑的任務竟然拋錯了(為什么要說幸運?)，于是馬上想到，之前那條 WAIT 狀態的線程可能是因為同樣的拋錯所以被中斷了，導致任務沒有繼續進行下去。

為什么說幸運?因為如果單線程的任務沒有拋錯的話，我可能很久都想不到是這個原因。

深入探究線程池的異常處理

工作上的問題到這里就找到原因了，之后的解決過程也十分簡單，這里就不提了。

但是疑問又來了，為什么使用線程池的時候，線程因異常被中斷卻沒有拋出任何信息呢?還有平時如果是在 main 函數里面的異常也會被拋出來，而不是像線程池這樣被吞掉。

如果子線程拋出了異常，線程池會如何進行處理呢?

我提交任務到線程池的方式是: threadPoolExecutor.submit(Runnbale task); ，后面了解到使用 execute() 方式提交任務會把異常日志給打出來，這里研究一下為什么使用 submit 提交任務，在任務中的異常會被“吞掉”。

對于 submit() 形式提交的任務，我們直接看源碼：

public Future<?> submit(Runnable task) { 
    if (task == null) throw new NullPointerException(); 
    // 被包裝成 RunnableFuture 對象，然后準備添加到工作隊列 
    RunnableFuture<Void> ftask = newTaskFor(task, null); 
    execute(ftask); 
    return ftask; 
} 

它會被線程池包裝成 RunnableFuture 對象，而最終它其實是一個 FutureTask 對象，在被添加到線程池的工作隊列，然后調用 start() 方法后， FutureTask 對象的 run() 方法開始運行，即本任務開始執行。

public void run() { 
    if (state != NEW || !UNSAFE.compareAndSwapObject(this,runnerOffset,null, Thread.currentThread())) 
        return; 
    try { 
        Callable<V> c = callable; 
        if (c != null && state == NEW) { 
            V result; 
            boolean ran; 
            try { 
                result = c.call(); 
                ran = true; 
            } catch (Throwable ex) { 
                // 捕獲子任務中的異常 
                result = null; 
                ran = false; 
                setException(ex); 
            } 
            if (ran) 
                set(result); 
        } 
    } finally { 
        runner = null; 
        int s = state; 
        if (s >= INTERRUPTING) 
            handlePossibleCancellationInterrupt(s); 
    } 
} 

在 FutureTask 對象的 run() 方法中，該任務拋出的異常被捕獲，然后在setException(ex); 方法中，拋出的異常會被放到 outcome 對象中，這個對象就是 submit() 方法會返回的 FutureTask 對象執行 get() 方法得到的結果。

但是在線程池中，并沒有獲取執行子線程的結果，所以異常也就沒有被拋出來，即被“吞掉”了。

這就是線程池的 submit() 方法提交任務沒有異常拋出的原因。

線程池自定義異常處理方法

在定義 ThreadFactory 的時候調用

setUncaughtExceptionHandler方法，自定義異常處理方法。例如：

ThreadFactory namedThreadFactory = new ThreadFactoryBuilder() 
                .setNameFormat("judge-pool-%d") 
                .setUncaughtExceptionHandler((thread, throwable)-> logger.error("ThreadPool {} got exception", thread,throwable)) 
                .build(); 

這樣，對于線程池中每條線程拋出的異常都會打下 error 日志，就不會看不到了。

后續

在修復了單個線程任務的異常之后，我繼續使用線程池進行重新統計業務，終于跑完了，也終于完成了這個任務。

事后我也叫三歪以后進公司一定要先邁出右腳進來，不然對寫代碼的風水影響很大。

小結：丙這個事故也給大家一個警示，使用線程池時需要注意，子線程的異常，如果沒有被捕獲就會丟失，可能會導致后期根據日志調試時無法找到原因。