代碼

重寫ThreadPoolExecutor 的 afterExecute方法

private class ExcaptionThreadPoolExecutor extends ThreadPoolExecutor{
        
        public ExcaptionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        
        }

        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            //存在錯(cuò)誤，打印到日志
            if(r instanceof FutureTask){
                try{
                    ((FutureTask) r).get();
                }catch (Throwable taskThr){
                    log.error("錯(cuò)誤：",taskThr);
                }
            }
            if(t != null){
                log.error("錯(cuò)誤：",t);
            }
        }
    }

原因

為什么要重寫ThreadPoolExecutor的afterExecute方法，因?yàn)樵诰€程池執(zhí)行run方法時(shí)，run方法被try塊包裹并將結(jié)果執(zhí)行到了afterExecute方法，而線程池的afterExecute方法并沒(méi)有任何實(shí)現(xiàn)

這是線程池內(nèi)執(zhí)行的run方法，在finally中執(zhí)行了afterExecute(task, thrown)方法：

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                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();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

但是如果執(zhí)行線程池時(shí)使用的是submit方法，而不是execute的話，情況又會(huì)有些不一樣了，那是因?yàn)閳?zhí)行submit的方法時(shí)，線程池將我們傳入的Runnable進(jìn)行了封裝，封裝成了FutureTask類

sumbit方法：

public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }

newTaskFor方法：

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

我們發(fā)現(xiàn)最后線程池執(zhí)行的其實(shí)是FutureTask類，而它對(duì)run又進(jìn)行了一次封裝

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 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);
        }
    }

也就是實(shí)際上我們的run方法是在這個(gè)run方法內(nèi)執(zhí)行的，而線程池的run方法執(zhí)行的實(shí)際上是RunnableFuture的run方法

從上面我們可以看出在發(fā)生catch時(shí)調(diào)用了setException方法，

protected void setException(Throwable t) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = t;
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
            finishCompletion();
        }
    }

也就是在setException中，我們發(fā)現(xiàn)它將Throwable賦值給了outcome變量

/** The result to return or exception to throw from get() */
private Object outcome; // non-volatile, protected by state reads/writes

這個(gè)變量在注釋中明確的指出了將用來(lái)存放返回值或者異常信息，在注釋中我們發(fā)現(xiàn)The result to return or exception to throw from get() ，使用get方法獲取信息

/**
 * @throws CancellationException {@inheritDoc}
 */
public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}

/**
 * Returns result or throws exception for completed task.
 *
 * @param s completed state value
 */
@SuppressWarnings("unchecked")
private V report(int s) throws ExecutionException {
    Object x = outcome;
    if (s == NORMAL)
        return (V)x;
    if (s >= CANCELLED)
        throw new CancellationException();
    throw new ExecutionException((Throwable)x);
}

至此我們拿到了信息，這也就解釋了代碼中為什么要判斷Runnable的實(shí)例是否是FutureTask類型，因?yàn)樵擃愋蜎](méi)有報(bào)錯(cuò)，它的錯(cuò)誤信息全都在實(shí)例中，不能用線程池的異常捕獲