线程池吞掉异常的case：源码阅读与解决方法

1. 问题背景

有一天给同事CR，看到一段这样的代码

try {
    for (param : params) {
        //并发处理，func无返回值
        ThreadPool.submit(func(param));
    }
} catch (Exception e) {
    log.info("func抛异常啦,参数是:{}", param)
}

我：你这段代码是利用并发降低RT对吧，如果func内部抛异常，你确定可以catch到吗

同事：可以啊， 为什么不可以（...

我：不如你run一把，在func mock一个异常出来试试

同事：我靠还真是

我：你可以用execute，改动比较小

同事：那么是为什么呢

2. 同事的例子

import java.util.concurrent.*;

public class ThreadPoolTest {

        public static void main(String[] args) throws Exception {
            ExecutorService executorService = new ThreadPoolExecutor(10, 20, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
            testExecute(executorService);
            Thread.sleep(2000);
            testSubmit1(executorService);
            Thread.sleep(2000);
            testSubmit2(executorService);
        }

        private static void testExecute(ExecutorService executorService) {
            executorService.execute(() -> {
                System.out.println("执行线程池execute方法");
                throw new RuntimeException("execute方法抛出异常");
            });
        }

        private static void testSubmit1(ExecutorService executorService) {
            executorService.submit(() -> {
                System.out.println("执行线程池submit方法1");
                throw new RuntimeException("submit方法1抛出异常");
            });
        }

        private static void testSubmit2(ExecutorService executorService) throws Exception {
            Future<Object> feature = executorService.submit(() -> {
                System.out.println("执行线程池submit方法2");
                throw new RuntimeException("submit方法2抛出异常");
            });
            feature.get();
        }
}

执行结果：

执行线程池execute方法
Exception in thread "pool-1-thread-1" java.lang.RuntimeException: execute方法抛出异常
	at ThreadPoolTest.lambda$testExecute$0(ThreadPoolTest.java:23)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
执行线程池submit方法1
执行线程池submit方法2
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.RuntimeException: submit方法2抛出异常
	at java.util.concurrent.FutureTask.report(FutureTask.java:122)
	at java.util.concurrent.FutureTask.get(FutureTask.java:192)
	at ThreadPoolTest.testSubmit2(ThreadPoolTest.java:39)
	at ThreadPoolTest.main(ThreadPoolTest.java:17)
Caused by: java.lang.RuntimeException: submit方法2抛出异常
	at ThreadPoolTest.lambda$testSubmit2$2(ThreadPoolTest.java:37)
	at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
	at java.util.concurrent.FutureTask.run(FutureTask.java)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

3. 原理分析

3.1 线程池包的继承结构

3.2 submit和execute方法的差异

3.2.1 execute

方法定义在最顶层的Executor接口，并且Executor接口有且仅有这一个方法

public interface Executor {

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the {@code Executor} implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution
     * @throws NullPointerException if command is null
     */
    void execute(Runnable command);
}

方法实现在ThreadPoolExecutor：

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        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);
    }

实际执行的过程，在worker（是runnable的实现类）的run方法，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);
                    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);
        }
    }

可以看到执行过程中，如果task.run();发生异常，没有catch处理，异常会层层向外抛出；最终进入finally块，执行processWorkerExit；

3.2.2 submit

submit方法定义在ExecutorService

public interface ExecutorService extends Executor {

    /**
     * Submits a value-returning task for execution and returns a
     * Future representing the pending results of the task. The
     * Future's {@code get} method will return the task's result upon
     * successful completion.
     *
     * <p>
     * If you would like to immediately block waiting
     * for a task, you can use constructions of the form
     * {@code result = exec.submit(aCallable).get();}
     *
     * <p>Note: The {@link Executors} class includes a set of methods
     * that can convert some other common closure-like objects,
     * for example, {@link java.security.PrivilegedAction} to
     * {@link Callable} form so they can be submitted.
     *
     * @param task the task to submit
     * @param <T> the type of the task's result
     * @return a Future representing pending completion of the task
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     * @throws NullPointerException if the task is null
     */
    <T> Future<T> submit(Callable<T> task);

    /**
     * Submits a Runnable task for execution and returns a Future
     * representing that task. The Future's {@code get} method will
     * return the given result upon successful completion.
     *
     * @param task the task to submit
     * @param result the result to return
     * @param <T> the type of the result
     * @return a Future representing pending completion of the task
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     * @throws NullPointerException if the task is null
     */
    <T> Future<T> submit(Runnable task, T result);

    /**
     * Submits a Runnable task for execution and returns a Future
     * representing that task. The Future's {@code get} method will
     * return {@code null} upon <em>successful</em> completion.
     *
     * @param task the task to submit
     * @return a Future representing pending completion of the task
     * @throws RejectedExecutionException if the task cannot be
     *         scheduled for execution
     * @throws NullPointerException if the task is null
     */
    Future<?> submit(Runnable task);
}

实现在AbstractExecutorService

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

可以看到这里创建了RunnableFuture（而不是基础的worker），顾名思义，RunnableFuture同时实现了Runnable和Future接口，也就意味着可以对该任务执行get操作，看看RunnableFuture的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);
        }
    }

catch块对方法异常做了处理，与执行结果一同在Future中暂存起来；submit()执行完毕后返回Future对象，执行future.get()会触发异常的抛出；

当然了，如果你只是执行了submit，没有获取future，异常就会“神奇地”消失。

参考：

Java线程池实现原理及其在美团业务中的实践 - 美团技术团队

https://zhuanlan.zhihu.com/p/651997713