转载

CompletableFuture使用大全，简单易懂

CompletableFuture是高级的多线程功能，支持自定义线程池和系统默认的线程池，是多线程，高并发里面，经常需要用到的比直接创建线程，要简单易用的方法。

本文主要内容如下：

CompletableFuture基本简述
API分类与记忆规律
创建CompletableFuture
取值与状态测试
控制CompletableFuture执行
CompletableFuture行为接续

CompletableFuture基本概述

CompletableFuture主要是用于异步调用，内部封装了线程池，可以将请求或者处理过程，进行异步处理。创建线程有3种方式，直接继承Thread、实现Runnable接口、实现Callable接口。以生活中的一个例子来说明异步行为：电饭煲蒸饭。

以前呀，都是大锅饭，放上米，放上水，然后需要不断地加柴火，人要看着火，具体什么时候煮熟，也得偶尔打开看看，看看开没开锅，煮没煮熟。这种就是没有任何通知方式，没有返回值的Runnable，只管煮饭，煮没煮熟需要自己判断。

一个老板发现了这个商机，说能不能做一个东西，不用人一直看着，自动就能把米饭做好，所以电饭煲就出现了。初代电饭煲的出现，算是解放了人力，再也不用看着火了，方便了很多，自己可以去做点其他的事情，热个牛奶，剪个鸡蛋什么的，但是至于饭什么时候熟，还得自己隔一段时间就得过去看一看。这就是Future的方式，虽然任务是异步执行的，但是要想获得这个结果，还得需要自己取。

时间继续推进，这个老板又有了新的想法，每隔一段时间，看看饭熟没熟还是有点浪费我看电视的时间，这个电饭煲能不能做好了，告诉我呢，这样我就直接来吃就行了。因此就有了这种可以预约、可以定时、可以保温的高级电饭煲。这个就对应着CompletableFuture，所有事情都是可以自动完成，即可以在完成之后，回调通知，也可以自己去等待。

Runnable就是没有返回结果的行为。
Callable是有返回结果的行为。
Future 异步封装Callable和Runnable，委托给线程池执行后，需要取回执行的结果
CompletableFuture 封装了Future，使其拥有了回调的功能，在某个行为完成之后，可以继续进行下一个动作。

示例代码：
public static void main(String[] args){
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
        System.out.println("电饭煲开始做饭");
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "白米饭";
    }).thenAccept(result -> {
        System.out.println("开始吃米饭");
    });

    System.out.println("我先去搞点牛奶和鸡蛋");
    future.join();
}

结果输出：
电饭煲开始做饭
我先去搞点牛奶和鸡蛋
开始吃米饭
复制代码

这样就可以一边等待米饭煮熟，一边去做其他事情。

API方法分类与记忆规律

CompletableFuture提供了方法大约有50多个，单纯一个个记忆，是很麻烦的，因此将其划分为以下几类：

创建类

completeFuture 可以用于创建默认返回值
runAsync 异步执行，无返回值
supplyAsync 异步执行，有返回值
anyOf 任意一个执行完成，就可以进行下一步动作
allOf 全部完成所有任务，才可以进行下一步任务

状态取值类

join 合并结果，等待
get 合并等待结果，可以增加超时时间;get和join区别，join只会抛出unchecked异常，get会返回具体的异常
getNow 如果结果计算完成或者异常了，则返回结果或异常；否则，返回valueIfAbsent的值
isCancelled
isCompletedExceptionally
isDone

控制类用于主动控制CompletableFuture的完成行为

complete
completeExceptionally
cancel

接续类 CompletableFuture 最重要的特性，没有这个的话，CompletableFuture就没意义了，用于注入回调行为。

thenApply, thenApplyAsync
thenAccept, thenAcceptAsync
thenRun, thenRunAsync
thenCombine, thenCombineAsync
thenAcceptBoth, thenAcceptBothAsync
runAfterBoth, runAfterBothAsync
applyToEither, applyToEitherAsync
acceptEither, acceptEitherAsync
runAfterEither, runAfterEitherAsync
thenCompose, thenComposeAsync
whenComplete, whenCompleteAsync
handle, handleAsync
exceptionally

上面的方法很多，我们没必要死记硬背，按照如下规律，会方便很多，记忆规则：

以Async结尾的方法，都是异步方法，对应的没有Async则是同步方法，一般都是一个异步方法对应一个同步方法。
以Async后缀结尾的方法，都有两个重载的方法，一个是使用内容的forkjoin线程池，一种是使用自定义线程池
以run开头的方法，其入口参数一定是无参的，并且没有返回值，类似于执行Runnable方法。
以supply开头的方法，入口也是没有参数的，但是有返回值
以Accept开头或者结尾的方法，入口参数是有参数，但是没有返回值
以Apply开头或者结尾的方法，入口有参数，有返回值
带有either后缀的方法，表示谁先完成就消费谁

记住上面几条，基本上就可以记住大部分的方法，剩下的其他方法，就可以单独记忆了。

创建CompletableFuture

创建CompletableFuture，其实就是将我们要煮的米饭，委托给电饭煲；要煮米饭，我们要准备这么几件事情，其一我们要制定制作米饭的方式，其二，我们要指定电饭煲。除此之外，我们也可以委托其他的事情，最后可以通过all或者any进行组合。

// 异步任务，无返回值，采用内部的forkjoin线程池
CompletableFuture c1 = CompletableFuture
.runAsync(()->{System.out.println("打开开关，开始制作，就不用管了")});

// 异步任务，无返回值，使用自定义的线程池
CompletableFuture c11 = CompletableFuture
.runAsync(()->{System.out.println("打开开关，开始制作，就不用管了")},newSingleThreadExecutor());

// 异步任务，有返回值，使用内部默认的线程池
CompletableFuture<String> c2 = CompletableFuture
.supplyAsync(()->{System.out.println("清洗米饭");return "干净的米饭";});

// 只要有一个完成，则完成，有一个抛出异常，则携带异常
CompletableFuture.anyOf(c1,c2);

// 必须等待所有的future全部完成才可以
CompletableFuture.allOf(c1,c2);

复制代码

取值与状态

常用的是下面的这几种
// 不抛出异常，阻塞的等待
future.join()
// 有异常则抛出异常，阻塞的等待，无限等待
future.get()
// 有异常则抛出异常，最长等待1个小时，一个小时之后，如果还没有数据，则异常。
future.get(1,TimeUnit.Hours)

复制代码

控制CompletableFuture执行

3种方式：
// 完成
future.complete("米饭");
// 异常
future.completeExceptionally();
// 取消，参数并没有实际意义，没任何卵用。
future.cancel(false);

复制代码

接续行为，用来描述上一件事做完之后，该做什么

接续方式有很多种，可以总结为一下三类：

CompletableFuture + (Runnable,Consumer,Function)
CompletableFuture + CompletableFuture
CompletableFuture + 处理结果

接续方式1

CompletableFuture future = CompletableFuture.supplyAsync(()->{
    System.out.println("投放和清洗制作米饭的材料");
    return "干净的没有新冠病毒的大米";
}).thenAcceptAsync(result->{
    System.out.println("通电，设定模式，开始煮米饭");
}).thenRunAsync(()->{
    System.out.println("米饭做好了，可以吃了");
})

复制代码

接续方式2

假如蒸米饭和、热牛奶、炒菜等已经是3个不同的CompletableFuture，可以使用接续方式2，将两个或者多个CompletableFuture组合在一起使用。

CompletableFuture rice = CompletableFuture.supplyAsync(()->{
    System.out.println("开始制作米饭，并获得煮熟的米饭");
    return "煮熟的米饭";
})

//煮米饭的同时呢，我又做了牛奶
CompletableFuture mike = CompletableFuture.supplyAsync(()->{
    System.out.println("开始热牛奶，并获得加热的牛奶");
    return "加热的牛奶";
});

// 我想两个都好了，才吃早饭，thenCombineAsync有入参，有返回值
mike.thenCombineAsync(rice,(m,r)->{
    System.out.println("我收获了早饭："+m+","+r);
    return m+r;
})
// 有入参，无返回值
mike.thenAcceptBothAsync(rice,(m,r)->{
   System.out.println("我收获了早饭："+m+","+r); 
});
// 无入参，入参会之
mike.runAfterBothAsync(rice,()->{
   System.out.println("我收获了早饭"); 
});

// 或者直接连接两个CompletableFuture
rice.thenComposeAsync(r->CompletableFuture.supplyAsync(()->{
    System.out.println("开始煮牛奶");
    System.out.println("同时开始煮米饭");
    return "mike";
}))

复制代码

接续方式3

如果我们只想做结果处理，也没有其他的接续动作，并且我们想要判断异常的情况，那么可以用接续方式3

whenCompleteAsync：处理完成或异常，无返回值
handleAsync：处理完成或异常，有返回值

CompletableFuture.supplyAsync(()->{
    System.out.println("开始蒸米饭");
    return "煮熟的米饭";
}).whenCompleteAsync((rich,exception)->{
    if (exception!=null){
        System.out.println("电饭煲坏了，米饭没做熟");
    }else{
        System.out.println("米饭熟了，可以吃了");
    }
})
// 有返回值
CompletableFuture.supplyAsync(()->{
    System.out.println("开始蒸米饭");
    return "煮熟的米饭";
}).handleAsync((rich,exception)->{
    if (exception!=null){
        System.out.println("电饭煲坏了，米饭没做熟");
    }else{
        System.out.println("米饭熟了，可以吃了");
    }
    return "准备冷一冷再吃米饭";
})

// 异常处理
CompletableFuture.supplyAsync(()->{
    System.out.println("开始蒸米饭");
    return "煮熟的米饭";
}).handleAsync((rich,exception)->{
    if (exception!=null){
        System.out.println("电饭煲坏了，米饭没做熟");
    }else{
        System.out.println("米饭熟了，可以吃了");
    }
    return "准备冷一冷再吃米饭";
}).exceptionally((exception)->{
    // 前置动作必须的是一个又返回值的操作，不能是那种返回值的那种
    return "";
});

复制代码

CompletableFuture用了之后，才觉得这个东西，确实好用一些，不经意间就做成了异步处理，而且支持自定义的线程池，如果结合stream，可以轻松地实现多线程并发处理。

List<CompletableFuture<YoutubeVideoEntity>> futures = subVideosList.stream()
        .map(item ->
                CompletableFuture.supplyAsync(() -> this.getRetry(item)
                        , ThreadPoolHolder.BG_CRAWLER_POOL)
        ).collect(Collectors.toList());

List<YoutubeVideoEntity> videoEntities = futures.stream().map(CompletableFuture::join)
.filter(item -> item != null && item.getVideoId() != null).collect(Collectors.toList());
复制代码

原文 https://juejin.im/post/5eeb38eb51882565cb498b47

正文到此结束