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

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

article 2024/11/16 9:57:09

文章目录

  • 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并不会及时感知异常，而是等所有任务都执行完之后才往下继续运行，那此处会有两种情况：

  1. 使用完allOf之后，还要去做流程编排，不去直接get的话，这种方式没有问题。
  2. 需要汇聚所有的子线程执行结果返回给主线程，这种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("执行后");

    }
}

运行完之后，我们再去看下执行日志

你会发现，因为子线程的睡眠时间为传进来的num值，当num=1时，触发告警，因此主线程在等待一秒中就会感知到异常，郑如日志打印的时间间隔为1秒钟。

此时再去想想：那此时抛出异常时，num为3,5执行状态时怎么样的，再去debug一下触发异常这块代码：

你会发现当num=1抛出异常的时候，其他两个线程的执行状态还是未完成，这就是和allOf这个方法最大的区别。

是否感觉只有自己试验过才会有更大的收获，今天就先分享到这，后面干货多多！！

感觉对您有所启发的话，记得帮忙点赞，收藏加关注，并且分享给需要的小伙伴！！加油！！


http://www.kler.cn/a/314168.html

相关文章：

  • 针对gitgitee的使用
  • 数据库的性能优化 -- SQL性能优化
  • 昆明华厦眼科医院举办中外专家眼科技术研讨会
  • 给阿里云OSS绑定域名并启用SSL
  • nginx proxy_pass中斜杠问题
  • python习题练习
  • 重修设计模式-结构型-组合模式
  • 网络丢包定位记录(三)
  • 海外大带宽服务器连接失败怎么办?
  • Antd框架中的Select组件placeholder不显示
  • [苍穹外卖]-11数据可视化接口开发
  • Qt 窗口事件机制
  • 分页查询，pageHelper, pagehelper-spring-boot-starter
  • C++ 策略技术中的算法策略
  • 握手传输 状态机序列检测（记忆科技笔试题）_2024年9月2日
  • 构建高可用和高防御力的云服务架构第一部分：深入解析DDoS高防（1/5）
  • 财富之眼用经济思维看清世界PDF高清下载
  • 【FastAPI】服务器使用SSE实现客户端之间的广播和点对点功能
  • 给新人的python笔记（一）
  • 深度学习基本概念详解
  • flink on k8s
  • 79篇vs13篇！本周中国学者发文量远超外国学者| NHANES数据库周报（8.28~9.3）
  • 执行matlab后进行RTL功能仿真check
  • 基于开源鸿蒙(OpenHarmony)的【智能家居综合应用】系统
  • 代理模式---静态代理和动态代理
  • JVM内存学习