Java 8 的异步利器：CompletableFuture源码解析（建议精读）

2023-08-04 591

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 实现了俩接口，本身是个class。这个是Future的实现类，使用 completionStage 接口去支持完成时触发的函数和操作。

completableFuture 是JDK1.8版本新引入的类。下面是这个类：

实现了俩接口，本身是个class。这个是Future的实现类，使用 completionStage 接口去支持完成时触发的函数和操作。

一个 completetableFuture 就代表了一个任务，他能用Future的方法，还能做一些之前说的 executorService 配合 futures 做不了的。

之前future需要等待isDone为true才能知道任务跑完了，或者就是用get方法调用的时候会出现阻塞，而使用 completableFuture 的使用就可以用then，when等等操作来防止以上的阻塞和轮询isDone的现象出现。

1.创建CompletableFuture直接new对象。

一个 completableFuture 对象代表着一个任务，这个对象能跟这个任务产生联系。

下面用的 complete 方法意思就是这个任务完成了需要返回的结果，然后用 get() 方法可以获取到。

2.JDK1.8使用的接口类。

在本文的 CompletableFuture 中大量地使用了这些函数式接口。

注：这些声明大量应用于方法的入参中，像 thenApply 和 thenAccept 这俩就是一个用Function一个用Consumer

而lambda函数正好是可以作为这些接口的实现。例如 s->{return 1;} 这个就相当于一个Function。因为有入参和返回结果。

(1)Function

(2)Consumer

对于前面有Bi的就是这样的，BiConsumer就是两个参数的。

(3)Predicate这个接口声明是一个入参，返回一个boolean。

(4)supplier

3.下面是这个类的静态方法

带有Async就是异步执行的意思、也是一个 completableFuture 对象代表着一个任务这个原则。

这种异步方法都可以指定一个线程池作为任务的运行环境，如果没有指定就会使用 ForkJoinPool 线程池来执行

(1) supplyAsync&runAsync 的使用例子。

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService executorService = Executors.newCachedThreadPool();
    executorService.submit(new Callable<Object>() {
        @Override
        public Object call() throws Exception {
            System.out.println("executorService 是否为守护线程 :" + Thread.currentThread().isDaemon());
            return null;
        }
    });
    final CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("this is lambda supplyAsync");
        System.out.println("supplyAsync 是否为守护线程 " + Thread.currentThread().isDaemon());
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("this lambda is executed by forkJoinPool");
        return "result1";
    });
    final CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("this is task with executor");
System.out.println("supplyAsync 使用executorService 时是否为守护线程 : " + Thread.currentThread().isDaemon());
        return "result2";
    }, executorService);
    System.out.println(completableFuture.get());
    System.out.println(future.get());
    executorService.shutdown();
}

这些任务中带有supply是持有返回值的，run是void返回值的，在玩supply时发现一个问题：如果使用supplyAsync任务时不使用任务的返回值，即 不用get方法阻塞主线程会导致任务执行中断。

注：跟get方法无关，后面有答案

然后我开始探索是否是只有 supplyAsync 是这样。我测试了 runAsync 发现也是这样。

下图为与 supplyAsync 任务执行不全面一样的问题，我甚至测试了将lambda换成runnable发现无济于事。

答案：

造成这个原因是因为Daemon。因为 completableFuture 这套使用异步任务的操作都是创建成了守护线程，那么我们没有调用get方法不阻塞这个主线程的时候。主线程执行完毕，所有线程执行完毕就会导致一个问题，就是守护线程退出。

那么我们没有执行的代码就是因为主线程不再跑任务而关闭导致的，可能这个不叫问题，因为在开发中我们主线程常常是一直开着的。但是这个小问题同样让我想了好久。

下面我们开一个非守护线程，可以看到程序执行顺利。

下面证实守护线程在其他非守护线程全部退出的情况下不继续执行。

final CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
    System.out.println("this is lambda supplyAsync");
    System.out.println("supplyAsync 是否为守护线程 " + Thread.currentThread().isDaemon());
    try {
        TimeUnit.SECONDS.sleep(1);
        try(BufferedWriter writer = new BufferedWriter
                (new OutputStreamWriter(new FileOutputStream(new File("/Users/zhangyong/Desktop/temp/out.txt"))))){
            writer.write("this is completableFuture daemon test");
        }catch (Exception e){
            System.out.println("exception find");
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("this lambda is executed by forkJoinPool");
    return "result1";
});

这个代码就是操作本地文件，并且sleep了一秒。其他线程就一句控制台输出的代码，最终的结果是文件没有任何变化。

当我把主线程 sleep 5 秒时，本地文件会写入一句 this is completableFuture daemon test 验证成功。

(2)allOf&anyOf

这两个方法的入参是一个 completableFuture 组、allOf就是所有任务都完成时返回，但是是个Void的返回值。

anyOf是当入参的 completableFuture 组中有一个任务执行完毕就返回，返回结果是第一个完成的任务的结果。

public static void otherStaticMethod() throws ExecutionException, InterruptedException {
        final CompletableFuture<String> futureOne = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                System.out.println("futureOne InterruptedException");
            }
            return "futureOneResult";
        });
        final CompletableFuture<String> futureTwo = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(6000);
            } catch (InterruptedException e) {
                System.out.println("futureTwo InterruptedException");
            }
            return "futureTwoResult";
        });
        CompletableFuture future = CompletableFuture.allOf(futureOne, futureTwo);
        System.out.println(future.get());
//        CompletableFuture completableFuture = CompletableFuture.anyOf(futureOne, futureTwo);
//        System.out.println(completableFuture.get());
    }