线程池的异常处理机制

先上两段代码：

ExecutorService threadPool = Executors.newFixedThreadPool(1);
threadPool.execute(() -> {
    System.out.println("execute");
    Object obj = null;
    System.out.println(obj.toString());
});
threadPool.shutdown();
System.out.println("**********");

ExecutorService threadPool = Executors.newFixedThreadPool(1);
threadPool.submit(() -> {
    System.out.println("submit");
    Object obj = null;
    System.out.println(obj.toString());
});
threadPool.shutdown();
System.out.println("**********");

运行结果发现，execute执行的时候抛出了预期的NullPointerException异常，而submit执行时什么都没有！

三个字：跟源码。

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

​ 使用submit提交任务时，会先把task包装成FutureTask对象，然后才去调用execute来执行任务。继续看execute：

public void execute(Runnable command) {
        // ...
        // 如果线程个数消息核心线程数则新增处理线程处理
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 如果当前线程个数已经达到核心线程数则任务放入队列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        // 尝试新增处理线程进行处理
        else if (!addWorker(command, false))
            reject(command);// 新增失败则调用拒绝策略
    }

​ 这里是一个典型的生产者，通过addWorker()方法将任务包装成Worker提交到workQueue中。来看Worker代码：

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    final Thread thread;
    Runnable firstTask;
    volatile long completedTasks;

    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    public void run() {
        runWorker(this);
    }
    
    //……
}

​ 忽略一些细节的地方，Worker就是一个Runnable，run()方法调用的是runWorker()：

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);  //可重写此方法，ThreadPoolExecutor类中
                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);  //可重写此方法，ThreadPoolExecutor类中
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

---

​ execute的方式提交时，这里的task就是Runnable（上面代码有），因此就会直接抛出，最终JVM会去调用Thread#dispatchUncaughtException方法:

/**
 * Dispatch an uncaught exception to the handler. This method is
 * intended to be called only by the JVM.
 */
private void dispatchUncaughtException(Throwable e) {
    getUncaughtExceptionHandler().uncaughtException(this, e);
}

public UncaughtExceptionHandler getUncaughtExceptionHandler() {
    return uncaughtExceptionHandler != null ?
        uncaughtExceptionHandler : group;
}

一般没有去设置uncaughtExceptionHandler 时，getUncaughtExceptionHandler()返回的是group，即ThreadGroup实例，来看ThreadGroup#uncaughtException：

public void uncaughtException(Thread t, Throwable e) {
    if (parent != null) {
        parent.uncaughtException(t, e);
    } else {
        Thread.UncaughtExceptionHandler ueh =
            Thread.getDefaultUncaughtExceptionHandler();
        if (ueh != null) {
            ueh.uncaughtException(t, e);
        } else if (!(e instanceof ThreadDeath)) {
            System.err.print("Exception in thread \""
                             + t.getName() + "\" ");
            e.printStackTrace(System.err);
        }
    }
}

​ 可以看到，先看有没有parent，没有则通过Thread.getDefaultUncaughtExceptionHandler()获取UncaughtExceptionHandler来处理，没有的话才将异常打印到System.err里面。

​ 那么submit的方式提交时，是怎么的情况呢？runWorker()中task是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 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);
    }
}

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

可以看出，对于异常并没有抛出，而是把异常给了outcome，这样在调用get()方法的时候，才会拿到这个异常。

​ OK，到此为止已经解释了最开始的两段代码运行结果不同的原因了。

在写代码时对于线程池、包括线程的异常处理最好是直接try/catch。 除此之外还有其它的方式，但是不推荐：

    ExecutorService threadPool = Executors.newFixedThreadPool(1, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                @Override
                public void uncaughtException(Thread t, Throwable e) {
                    System.out.println("uncaughtException");
                    System.out.println(e);
                }
            });
            return t;
        }
    });

    threadPool.execute(() -> {
        System.out.println("execute");
        Object obj = null;
        System.out.println(obj.toString());
    });

    threadPool.shutdown();
    System.out.println("**********");
}