ExecutorService的invokeAll方法无法抛出异常

最近在线上发现一个多线程查询有问题，始终查询不到数据，但是查看线上日志又未发现报错日志，经过排查发现是由于invokeAll方法将抛出的异常进行二次转换成ExecutionException，但是又没有对该异常进行处理导致

invokeAll方法介绍(不想看的可以直接跳过)

invokeAll方法接收一个Callable集合，并返回一个Future对象的列表，它可以安排多个任务在线程池中执行，然后等待这些任务都执行完成。invokeAll方法会阻塞当前线程，直到所有任务都完成，或者超时(如果提供)，或者线程被中断，此时它将抛出InterruptedException异常。它允许任务之间存在依赖关系，即一个任务可以依赖于其他任务的结果。它还允许客户端获取每个任务的执行结果，这是通过返回的Future对象列表来实现的

验证

public class InvokeAllMethodTest {

    public static void main(String[] args) {
        // 自定义线程池
        ThreadPoolExecutor threadPoolExecutor =
            new ThreadPoolExecutor(10, 20, 60, TimeUnit.SECONDS, new SynchronousQueue<>(), Executors.defaultThreadFactory(), new AbortPolicy());


        // 创建要执行的任务
        List<Callable<Integer>> futures = Lists.newArrayList();
        futures.add(() -> {
            // 执行的业务逻辑出现异常
            if (true) {
                throw new RuntimeException("发生业务逻辑异常了");
            }
            return 1;
        });
        try {
            threadPoolExecutor.invokeAll(futures);
            System.out.println("=============运行结束===============");
        } catch (InterruptedException e) {
            System.out.println("=========发生中断异常了============");
            throw new RuntimeException(e);
        }catch (Exception e){
            System.out.println("=========发生其他异常了============");
            throw new RuntimeException(e);
        }
    }
}

运行结果截图

0.jpg

1.jpg

可以看出代码中已经明确的抛出了指定异常，但是在调用invokeAll方法后却并未将异常抛出

解释

进入AbstractExecutorService类的invokeAll方法

    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
        if (tasks == null)
            throw new NullPointerException();
        // 创建一个任务集合
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        // 任务是否完成的标识
        boolean done = false;
        try {
            for (Callable<T> t : tasks) {
                // 把Callable线程封装成了FutureTask
                RunnableFuture<T> f = newTaskFor(t);
                futures.add(f);
                // 执行任务
                execute(f);
            }
            for (int i = 0, size = futures.size(); i < size; i++) {
                Future<T> f = futures.get(i);
                if (!f.isDone()) {
                    try {
                        /* 获取线程执行的结果,
                         这里也能保证所有的线程都执行完才会返回
                         这里的get方法也就是为什么异常被吞的真正的罪魁祸首，将我们抛出的异常转成了ExecutionException 异常
                         前面讲到f已经是FutureTask类型了,所以直接找到他的get方法
                        */
                        f.get();
                    } catch (CancellationException ignore) {
                    } catch (ExecutionException ignore) {
                    }
                }
            }
            done = true;
            return futures;
        } finally {
            if (!done)
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
        }
    }

下面通过debug查看源码的执行过程：

2.jpg

3.jpg

4.jpg

5.jpg

解决方案

如果想要批量执行任务，这里推荐使用CompletableFuture ,这里不做详细介绍，具体可以参考博客：CompletableFuture用法详解

改造后的代码：

public class InvokeAllMethodTest {

    public static void main(String[] args) {
        // 自定义线程池
        ThreadPoolExecutor threadPoolExecutor =
            new ThreadPoolExecutor(10, 20, 60, TimeUnit.SECONDS, new SynchronousQueue<>(), Executors.defaultThreadFactory(), new AbortPolicy());


        // 创建要执行的任务
        List<CompletableFuture<Integer>> tasks = Lists.newArrayList();
        tasks.add(CompletableFuture.supplyAsync(()->{
            System.out.println("===========任务1正常执行===========");
            return 1;
        }));
        tasks.add(CompletableFuture.supplyAsync(()->{
            // 执行的业务逻辑出现异常
            if (true) {
                throw new RuntimeException("任务2发生业务逻辑异常了");
            }
            return 1;
        }));


        // 执行，由于allOf方法的入参是可变参数，因此将参数通过toArray转成数组，通过join执行
        CompletableFuture.allOf(tasks.toArray(new CompletableFuture[0])).join();

    }
}

执行结果：

6.jpg

ExecutorService的invokeAll方法无法抛出异常