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

CompletableFuture的allOf一定不要乱用！血泪史复盘

article 2024/9/22 4:13:25

文章目录

1. 到底遇到了什么问题？
2. CountDownLatch搞起？
3. allOf里面的坑
4. 优化建议：

1. 到底遇到了什么问题？

最近看到组里面的同学遇到了这样的业务场景：

主线程需要异步并发调用多个接口，并且主线程需要等待线程全部执行完成之后，返回执行结果，业务流程如下：

这里面有一个注意点：一旦异步线程有一个失败，我主线程就不等待了，这种需求很常见，比如多线程拼装对象数据等等。那我们如何解决呢？

一定要坚持看完！细节往往决定成败，写代码也一样！！收获多多
在这里插入图片描述

2. CountDownLatch搞起？

此时有可能有可能第一时间会想到CountDownLatch，但是CountDownLatch本质内部是采用计数器，主线程一直调用await()阻塞等待，需要等待所有子线程执行完之后（不管成功或者失败）主线程才会往下执行，且主线程和子线程无法传递异常，并且需要注意异常的抛出。

这块可以参考之前写的CountDownLatch文章：

关于CountDownLatch的底层源码和闭坑指南，只看这一篇就够了！！

这里面有常见的坑一栏，可以重点看下，保你必有收获，所以CountDownLatch并不是最优方案。

这块很多同学会选用CompletableFuture来实现，因为CompletableFuture底层有丰富的任务编排和链式调用，并且此时肯定会有很多同学说CompletableFuture.allOf可以轻松实现。

此时如果你不自己去试试的话，八股文一背，此时你就被带偏了！！我们可以去看一下，在接下来会给出一个allOf的使用场景，把这个问题重现一下：

3. allOf里面的坑

场景：当有一批任务交给线程池执行，我们需要获取所有线程的返回结果。

首先定义一个并发执行器类：

public class CompletableFutureEngine {

    private final static ExecutorService executorService = Executors.newFixedThreadPool(4);

    /**
     * 创建并行任务并执行
     *
     * @param list            数据源
     * @param function        API调用逻辑
     * @param exceptionHandle 异常处理逻辑
     * @return 处理结果列表
     */
    public static <S, T> List<T> parallelFutureJoin(Collection<S> list, Function<S, T> function, Consumer<Throwable> exceptionHandle) {

        List<CompletableFuture<T>> completableFutures = list.stream()
                .map(s -> CompletableFuture.supplyAsync(() -> function.apply(s)))
                .collect(Collectors.toList());

        List<T> results = new ArrayList<>();
        try {
            CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0])).join();
            for (CompletableFuture<T> completableFuture : completableFutures) {
                results.add(completableFuture.get());
            }
        } catch (Exception e) {
            if (e instanceof CompletionException) {
                if (e.getCause() != null) {
                    exceptionHandle.accept(e.getCause());
                }
            }
        }
        return results;
    }

}

调用方法：

public class EngineDemo {

    private static void sleep(long sleepTime) {
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void currentDate(String str) {
        // 创建一个Date对象，它包含了当前时间
        Date now = new Date();
        // 创建一个SimpleDateFormat对象，用于指定输出格式
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        // 使用format方法将Date对象格式化为字符串
        String currentTime = dateFormat.format(now);
        // 打印当前时间
        System.out.println(str + "是: " + currentTime);
    }

    public static void main(String[] args) {
        currentDate("执行前");
        List<Integer> numList = CompletableFutureEngine.parallelFutureJoin(Arrays.asList(1, 3, 5),
            num -> {
                sleep(num * 1000);
                if (num == 1) {
                    throw new BusinessException("心别太大");
                }
                return num;
            }
            , e -> {
                if (e instanceof BusinessException) {
                    System.out.println("BusinessException =" + e.getMessage());
                } else {
                    System.out.println("Exception entrance");
                }
            }
        );
        System.out.println(numList);
        currentDate("执行后");

    }
}

此时，你会发现，程序的执行时间为5秒钟，感觉似乎和想的不一样，因为在代码中，是根据num决定的休眠时间，因此在第一印象中，应该是第一秒执行完就会抛出异常。尴尬了，现在从日志看程序的执行时间是5秒，那我们看下原因：走起，那我们debug下代码：

你会发现当程序抛出异常的时候，发现传进来的三个CompletableFuture，不管是成功还是失败都执行完了。

这也就说明了当抛出异常后，allOf并不会及时感知异常，而是等所有任务都执行完之后才往下继续运行，那此处会有两种情况：

使用完allOf之后，还要去做流程编排，不去直接get的话，这种方式没有问题。
需要汇聚所有的子线程执行结果返回给主线程，这种allOf可能效率会低，因为需要等待所有的子线程执行完才会去返回最终的结果，那如果遇到主要一个子线程失败，执行就失败，就会导致执行时间短的失败了，但是有个任务执行时间很长，返回时间也会变长，从而导致主线程等待的时间变长.

那针对于第二种情况应该如何优化呢！！！继续往下看！！！

4. 优化建议：

可以用CompletableFutures.stream().map(CompletableFuture::join).collect(Collectors.toList()) 来代替 allOf。这个方法只要有一个子线程出现异常，主线程就会感知到异常，不用等待其他线程执行完。

优化后的并发执行器如下：

public class CompletableFutureEngine2 {

    /**
     * 创建并行任务并执行
     *
     * @param list            数据源
     * @param function        API调用逻辑
     * @return 处理结果列表
     */
    public static <S, T> List<T> parallelFutureJoin(Collection<S> list, Function<S, T> function, Consumer<Throwable> exceptionHandle) {

        List<CompletableFuture<T>> completableFutures = list.stream()
                .map(s -> CompletableFuture.supplyAsync(() -> function.apply(s)))
                .collect(Collectors.toList());

        List<T> results = null;
        try {
            results = completableFutures.stream().map(CompletableFuture::join).collect(Collectors.toList());
        } catch (Exception e) {
            if (e instanceof CompletionException) {
                if (e.getCause() != null) {
                    exceptionHandle.accept(e.getCause());
                }
            }
        }
        return results;
    }
}

调用demo如下：

public class EngineDemo {

    private static void sleep(long sleepTime) {
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void currentDate(String str) {
        // 创建一个Date对象，它包含了当前时间
        Date now = new Date();
        // 创建一个SimpleDateFormat对象，用于指定输出格式
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        // 使用format方法将Date对象格式化为字符串
        String currentTime = dateFormat.format(now);
        // 打印当前时间
        System.out.println(str + "是: " + currentTime);
    }

    public static void main(String[] args) {
        currentDate("执行前");
        List<Integer> numList = CompletableFutureEngine2.parallelFutureJoin(Arrays.asList(1, 3, 5),
            num -> {
                sleep(num * 1000);
                if (num == 1) {
                    throw new BusinessException("心别太大");
                }
                return num;
            }
            , e -> {
                if (e instanceof BusinessException) {
                    System.out.println("BusinessException =" + e.getMessage());
                } else {
                    System.out.println("Exception entrance");
                }
            }
        );
        System.out.println(numList);
        currentDate("执行后");

    }
}