当前位置：首页 > news >正文

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

news 来源：原创 2024/6/26 21:56:28

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 interruptsboolean 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 interruptif ((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 interruptsint s = state;if (s >= INTERRUPTING)handlePossibleCancellationInterrupt(s);}}

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

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

参考：

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

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