Java ExecutorService線程池中的小坑——關(guān)于線程池中拋出的異常處理

wow_worktile 發(fā)布于2019-08-14 11:50 / 2637人閱讀

摘要：先看寫的簡略的代碼線程池中發(fā)現(xiàn)異常，被中斷線程池中發(fā)現(xiàn)異常，被中斷我這是一個訂單處理流程，主要用到了一個方法，就是。好了，以上就是對線程池異常捕捉的一個記錄。

開發(fā)自己的項目有一段時間了，因為是個長時間跑的服務(wù)器端程序，所以異常處理顯得尤為重要。
對于異常的抓取和日志（狹義上的日志）的分析一點都不能落下。

我們使用了Java自帶的Executor模塊，我只是稍微看了下Executors當(dāng)中三個線程池的實現(xiàn)(策略為：Fixed, Cached, Schedule)，其實光看名字就可以了解各自的一些策略信息。OK，這一次我需要一種策略合并Fixed和Cached的兩種特點的自定義Executor。其實很簡單，給Cached設(shè)置一個上線就是了。注意他們的同步隊列使用的不同，用LinkedBlockingQueue是個不錯的選擇，至于BlockingQueue的實現(xiàn)可以自行谷歌（以后再記吧）。

先看寫的簡略的代碼

package com.zjseek.recharge.core;

import com.zjseek.recharge.exception.SKErrorCode;
import com.zjseek.recharge.exception.SKOrderState;
import com.zjseek.recharge.model.OrderModel;
import com.zjseek.recharge.service.OrderService;
import org.apache.log4j.Logger;

import java.sql.Timestamp;
import java.util.concurrent.*;

/**
 * Created by geminiwen on 14-6-28.
 */
public class OrderExceptionThreadExecutor extends ThreadPoolExecutor {
    private Logger logger = Logger.getLogger(OrderExceptionThreadExecutor.class);
    private OrderService orderService;

    public OrderExceptionThreadExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        init();
    }

    public OrderExceptionThreadExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        init();
    }

    public OrderExceptionThreadExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        init();
    }

    public OrderExceptionThreadExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        init();
    }

    private void init() {
        this.orderService = new OrderService();
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        Future f = (Future) r;
        try {
            f.get();
        } catch (InterruptedException e) {
            logger.error("線程池中發(fā)現(xiàn)異常，被中斷", e);
        } catch (ExecutionException e) {
            logger.error("線程池中發(fā)現(xiàn)異常，被中斷", e);
        }

    }
}

我這是一個訂單處理流程，主要用到了一個protected方法，就是afterExecute。一看這個函數(shù)的樣子，想當(dāng)然的以為如果線程池中出了問題，異常自然回在第二個參數(shù)t中傳過來。
也許的確是這樣的，但是這里有一個區(qū)別。
我們知道ExecutorServcie中執(zhí)行一個Runnable有兩個方法，兩個分別是

public void execute(Runnable command);
public  Future submit(Runnable task, T result);

別看接受的參數(shù)差不多，其實submit最后是調(diào)用的execute的，而且在調(diào)用execute前，對task進行了一次封裝，變成了RunnableFuture（它是接口，繼承了Runnable和Future實際是一個實現(xiàn)類FutureTask）。

OK，對于實際操作Runnable的不同，暫時說到這，看下execute方法做了什么事
execute方法對進來的Runnable又包裝成了worker然后進入runWorker
runWorker方法中有這么幾行

try {
    beforeExecute(wt, task);
    Throwable thrown = null;
    try {
       task.run();
    } catch (RuntimeException x) {
       thrown = x; throw x;
    } catch (Error x) {
       thrown = x; throw x;
    } catch (Throwable x) {
       thrown = x; throw new Error(x);
    } finally {
       afterExecute(task, thrown);
    }
} finally {
    task = null;
    w.completedTasks++;
    w.unlock();
}

好了，到了最關(guān)鍵的afterExecute這個步驟，我滿心以為這里所有的異常都會通過thrown傳遞進來，看來我還是太年輕了，之前我們分析過，這個Runnable已經(jīng)被submit封裝成了FutureTask，那么這個task.run()除了我們自己定義的run任務(wù)之外，到底還干了啥呢？

public void run() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        Callable 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 must be non-null until state is settled to
        // prevent concurrent calls to run()
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

OK，這段源碼摘自FutureTask中的run方法，實際我們自己定義的任務(wù)已經(jīng)變成了Callable：

   public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

從它的構(gòu)造函數(shù)就可以看出來。

然后我們在上面實際運行task的地方其實是c.call()這一句。

  result = c.call();

我們寫的任務(wù)全部在這句代碼里面執(zhí)行完畢了，看看外面都wrap了啥？ OK 我們所有的Throwable全部已經(jīng)被setException吃掉了，怎么還會拋出到外面那層的execute中呢？
所以我之前實驗的時候，在submit中提交任務(wù)無論任務(wù)怎么拋異常，在afterExecute中的第二個參數(shù)是取不到的，原因就在這。

再回頭看看針對submit改造的函數(shù)

protected void afterExecute(Runnable r, Throwable t) {
    super.afterExecute(r, t);
    Future f = (Future) r;
    try {
        f.get();
    } catch (InterruptedException e) {
        logger.error("線程池中發(fā)現(xiàn)異常，被中斷", e);
    } catch (ExecutionException e) {
        logger.error("線程池中發(fā)現(xiàn)異常，被中斷", e);
    }

}

當(dāng)然，這里已經(jīng)默認r是實現(xiàn)Future接口了。通過FutureTask的get方法，能把剛剛setException中的異常給拋出來，這樣我們就能真的拿到這些異常了。

結(jié)論

如果我們關(guān)心線程池執(zhí)行的結(jié)果，則需要使用submit來提交task，那么在afterExecute中對異常的處理也需要通過Future接口調(diào)用get方法去取結(jié)果，才能拿到異常，如果我們不關(guān)心這個任務(wù)的結(jié)果，可以直接使用ExecutorService中的execute方法（實際是繼承Executor接口）來直接去執(zhí)行任務(wù)，這樣的話，我們的Runnable沒有經(jīng)過多余的封裝，在runWorker中得到的異常也直接能在afterExecute中捕捉。

好了，以上就是對線程池異常捕捉的一個記錄。想想應(yīng)該不難，今天也是偶然機會看到的。今天在開發(fā)中碰到PHP鎖的問題，頭疼死了。