Java并发编程—CompletableFuture的介绍和使用

news2024/11/8 11:19:27

在博主上一篇博客介绍中，Java并发编程—java异步Future的迭代过程_小魏快起床的博客-CSDN博客，这里面给大家分析了Future的使用过程和一些存在的问题，那么针对里面出现的阻塞问题，博主将在这一篇文章给大家介绍清楚

🍏一、认识新的类CompletableFuture

🍄1、CompletableFuture类在java源代码中的样子：

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {	}

🍄2、CompletableFuture类的架构图

🍄3、再回顾一下博主上一篇博客中讲的FutureTask类结构图：

🚗 综上:这个新的CompletableFuture类实现了和FutureTask一样的Future类，所以CompletableFuture类里面有FutureTask关于异步任务的这些特性。同时，CompletableFuture还实现了CompletionStage接口，而真正解决之前的阻塞问题的就是咱们CompletionStage接口，而CompletableFuture进行了一个实现，所以就给大家讲解的是CompletableFuture类；

🍏 二、CompletableFuture和CompletionStage源码分别介绍

🍄 1、接口CompletionStage：

①、CompletionStage代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段。

②、一个阶段的计算执行可以是一个Function，Consumer或者Runnable。

③、一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发；

🚗 总结：代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段，类似Linux系统的管道分隔符传参数，说人话就是一阶段完成了，需要做二阶段，这个时候二阶段可能会依靠一阶段的结果来执行一些业务。

🍄 2、类CompletableFuture：

CompletableFuture是对Future的扩展和增强。CompletableFuture实现了Future接口，并在此基础上进行了丰富的扩展，完美弥补了Future的局限性，同时CompletableFuture实现了对任务编排的能力。借助这项能力，可以轻松地组织不同任务的运行顺序、规则以及方式。从某种程度上说，这项能力是它的核心能力。而在以往，虽然通过CountDownLatch等工具类也可以实现任务的编排，但需要复杂的逻辑处理，不仅耗费精力且难以维护。

🚗 总结：CompletableFuture牛逼就完事儿，FutureTask可以做的这个都可以做，并且还可以做一些Future不能做的事；

🍏 三、CompletableFuture的方法介绍

介绍一下CompletableFuture里面的方法：

runAsync 创建无返回值的异步任务
supplyAsync 创建有返回值的异步任务

代码案例：

public static void main(String[] args) throws ExecutionException, InterruptedException {

        // 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                1,
                1,
                0,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(1)
        );

        //第一个基本方法runAsync
        CompletableFuture<Void> async1 = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "执行");
        });

        //第二个基本方法runAsync，带线程池的
        CompletableFuture<Void> async2 = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "执行");
        }, pool);

        //第三个基本方法supplyAsync
        CompletableFuture<Integer> async3 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "执行");
            return 11;
        });
        System.out.println(async3.get());

        //第四个基本方法supplyAsync
        CompletableFuture<Integer> async4 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t" + "执行");
            return 12;
        }, pool);
        System.out.println(async4.get());

        //关闭线程池
        pool.shutdown();
    }

🚗 总结：在上面的代码中，每个方法都有2种使用方法，无非就是携带线程池的区别，如果不提供自定义线程池的话，方法内部会用默认的线程池；

分别执行的效果如下，有数字的是有返回值的，工作中用哪种，取决于实际应用场景：

以上是对这个类和一些方法的初步解析

🍏 四、用CompletableFuompleture做之前的事儿

之前博主的想法是在主线程中开启一个子线程，然后子线程进行一个计算，然后返回结果值，那么用CompletableFuompleture来做一些试试：

//1、创建一个有返回值的异步任务
        CompletableFuture<Integer> async = CompletableFuture.supplyAsync(() -> {
            try {
                //延迟2s
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 11;
        });
        System.out.println("主线程正常执行");
        //获取异步任务的返回值
        System.out.println(async.get());
        System.out.println("结束");

效果如下：

分析：这样执行是没啥问题的，但是还是出现了阻塞的问题，明明都说了换类，都说了这个好用，为啥还会出现阻塞呢，这个显然是我们使用错误了，接下来就给大家演示正确的使用方法

🍏 五、CompletableFuompleture正确的使用方法：

博主想做这么一件事儿，开启一个异步任务，第一步计算1+2的值，然后将这个值再提供给下一步做+3的操作，然后输出最后的结果，实现一个多线程异步任务编排：

public static void main(String[] args) throws Exception {
        //1、创建一个线程池
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                1,
                1,
                0,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(1)
        );

        //2、创建异步编排任务
        CompletableFuture.supplyAsync(() -> {
            //延迟2s
            try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { }
            //做一下加法然后返回结果
            return 1 + 2;
        }, pool).thenApply(e -> {
            //根据上一步的结果+3
            return e + 3;
        }).whenComplete((v, f) -> {
            //判断是否有异常，如果没有就输出结果值
            if (f == null) {
                System.out.println("结果值：" + v);
            }
        }).exceptionally(e -> {
            //异常输出，whenComplete里面的f不为null时触发
            e.printStackTrace();
            return null;
        });

        //3、主线程执行输出
        System.out.println("输出main线程");

        //4、关闭线程池
        pool.shutdown();
        }

效果如下：