关于html:异步编程利器CompletableFuture如何操作

前言
最近刚好应用CompeletableFuture优化了我的项目中的代码，所以跟大家一起学习CompletableFuture。

一个例子回顾 Future
因为CompletableFuture实现了Future接口，咱们先来回顾Future吧。

Future是Java5新加的一个接口，它提供了一种异步并行计算的性能。如果主线程须要执行一个很耗时的计算工作，咱们就能够通过future把这个工作放到异步线程中执行。主线程持续解决其余工作，解决实现后，再通过Future获取计算结果。

来看个简略例子吧，假如咱们有两个工作服务，一个查问用户根本信息，一个是查问用户勋章信息。如下，

public class UserInfoService {

public UserInfo getUserInfo(Long userId) throws InterruptedException {    Thread.sleep(300);//模仿调用耗时    return new UserInfo("666", "捡田螺的小男孩", 27); //个别是查数据库，或者近程调用返回的}

}

public class MedalService {

public MedalInfo getMedalInfo(long userId) throws InterruptedException {    Thread.sleep(500); //模仿调用耗时    return new MedalInfo("666", "守护勋章");}

}
接下来，咱们来演示下，在主线程中是如何应用Future来进行异步调用的。

public class FutureTest {

public static void main(String[] args) throws ExecutionException, InterruptedException {    ExecutorService executorService = Executors.newFixedThreadPool(10);    UserInfoService userInfoService = new UserInfoService();    MedalService medalService = new MedalService();    long userId =666L;    long startTime = System.currentTimeMillis();    //调用用户服务获取用户根本信息    FutureTask<UserInfo> userInfoFutureTask = new FutureTask<>(new Callable<UserInfo>() {        @Override        public UserInfo call() throws Exception {            return userInfoService.getUserInfo(userId);        }    });    executorService.submit(userInfoFutureTask);    Thread.sleep(300); //模仿主线程其它操作耗时    FutureTask<MedalInfo> medalInfoFutureTask = new FutureTask<>(new Callable<MedalInfo>() {        @Override        public MedalInfo call() throws Exception {            return medalService.getMedalInfo(userId);        }    });    executorService.submit(medalInfoFutureTask);    UserInfo userInfo = userInfoFutureTask.get();//获取个人信息后果    MedalInfo medalInfo = medalInfoFutureTask.get();//获取勋章信息后果    System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");}

}

运行后果:

总共用时806ms
如果咱们不应用Future进行并行异步调用，而是在主线程串行进行的话，耗时大概为300+500+300 = 1100 ms。能够发现，future+线程池异步配合，进步了程序的执行效率。

然而Future对于后果的获取，不是很敌对，只能通过阻塞或者轮询的形式失去工作的后果。

Future.get() 就是阻塞调用，在线程获取后果之前get办法会始终阻塞。
Future提供了一个isDone办法，能够在程序中轮询这个办法查问执行后果。
阻塞的形式和异步编程的设计理念相违反，而轮询的形式会消耗无谓的CPU资源。因而，JDK8设计出CompletableFuture。CompletableFuture提供了一种观察者模式相似的机制，能够让工作执行实现后告诉监听的一方。

一个例子走进CompletableFuture
咱们还是基于以上Future的例子，改用CompletableFuture 来实现

public class FutureTest {

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {    UserInfoService userInfoService = new UserInfoService();    MedalService medalService = new MedalService();    long userId =666L;    long startTime = System.currentTimeMillis();    //调用用户服务获取用户根本信息    CompletableFuture<UserInfo> completableUserInfoFuture = CompletableFuture.supplyAsync(() -> userInfoService.getUserInfo(userId));    Thread.sleep(300); //模仿主线程其它操作耗时    CompletableFuture<MedalInfo> completableMedalInfoFuture = CompletableFuture.supplyAsync(() -> medalService.getMedalInfo(userId));     UserInfo userInfo = completableUserInfoFuture.get(2,TimeUnit.SECONDS);//获取个人信息后果    MedalInfo medalInfo = completableMedalInfoFuture.get();//获取勋章信息后果    System.out.println("总共用时" + (System.currentTimeMillis() - startTime) + "ms");}

}
能够发现，应用CompletableFuture，代码简洁了很多。CompletableFuture的supplyAsyncwww.cungun.com办法，提供了异步执行的性能，线程池也不必独自创立了。实际上，它CompletableFuture应用了默认线程池是ForkJoinPool.commonPool。

CompletableFuture提供了几十种办法，辅助咱们的异步工作场景。这些办法包含创立异步工作、工作异步回调、多个工作组合解决等方面。咱们一起来学习吧

CompletableFuture应用场景
0c2266ac4dec5f69a317e54fb8c99831.jpeg

创立异步工作
CompletableFuture创立异步工作，个别有supplyAsync和runAsync两个办法

319a725a11f165a2abf16bca4cbe3deb.jpeg创立异步工作

supplyAsync执行CompletableFuture工作，反对返回值
runAsync执行CompletableFuture工作，没有返回值。
supplyAsync办法
//应用默认内置线程池ForkJoinPool.commonPool()，依据supplier构建执行工作
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
//自定义线程，依据supplier构建执行工作
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
runAsync办法
//应用默认内置线程池ForkJoinPool.commonPool()，依据runnable构建执行工作
public static CompletableFuture<Void> runAsync(Runnable runnable)
//自定义线程，依据runnable构建执行工作
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
实例代码如下：

public class FutureTest {

public static void main(String[] args) {    //能够自定义线程池    ExecutorService executor = Executors.newCachedThreadPool();    //runAsync的应用    CompletableFuture<Void> runFuture = CompletableFuture.runAsync(() -> System.out.println("run,关注公众号:捡田螺的小男孩"), executor);    //supplyAsync的应用    CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> {                System.out.print("supply,关注公众号:捡田螺的小男孩");                return "捡田螺的小男孩"; }, executor);    //runAsync的future没有返回值，输入null    System.out.println(runFuture.join());    //supplyAsync的future，有返回值    System.out.println(supplyFuture.join());    executor.shutdown(); // 线程池须要敞开}

}
//输入
run,关注公众号:捡田螺的小男孩
null
supply,关注公众号:捡田螺的小男孩捡田螺的小男孩

工作异步回调
f2c5d4dcf61128632a330b068085e567.jpeg

1. thenRun/thenRunAsync
   public CompletableFuture<Void> thenRun(Runnable action);
   public CompletableFuture<Void> thenRunAsync(Runnable action);
   CompletableFuture的thenRun办法，艰深点讲就是，做完第一个工作后，再做第二个工作。某个工作执行实现后，执行回调办法；然而前后两个工作没有参数传递，第二个工作也没有返回值

public class FutureThenRunTest {

public static void main(String[] args) throws ExecutionException, InterruptedException {    CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(            ()->{                System.out.println("先执行第一个CompletableFuture办法工作");                return "捡田螺的小男孩";            }    );    CompletableFuture thenRunFuture = orgFuture.thenRun(() -> {        System.out.println("接着执行第二个工作");    });    System.out.println(thenRunFuture.get());}

}
//输入
先执行第一个CompletableFuture办法工作
接着执行第二个工作
null
thenRun 和thenRunAsync有什么区别呢？能够看下源码哈：

private static final Executor asyncPool = useCommonPool ?

    ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();    public CompletableFuture<Void> thenRun(Runnable action) {    return uniRunStage(null, action);}public CompletableFuture<Void> thenRunAsync(Runnable action) {    return uniRunStage(asyncPool, action);}

如果你执行第一个工作的时候，传入了一个自定义线程池：

调用thenRun办法执行第二个工作时，则第二个工作和第一个工作是共用同一个线程池。
调用thenRunAsync执行第二个工作时，则第一个工作应用的是你本人传入的线程池，第二个工作应用的是ForkJoin线程池
TIPS: 前面介绍的thenAccept和thenAcceptAsync，thenApply和thenApplyAsync等，它们之间的区别也是这个哈。

2.thenAccept/thenAcceptAsync
CompletableFuture的thenAccept办法示意，第一个工作执行实现后，执行第二个回调办法工作，会将该工作的执行后果，作为游戏入参，传递到回调办法中，然而回调办法是没有返回值的。

public class FutureThenAcceptTest {

public static void main(String[] args) throws ExecutionException, InterruptedException {    CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(            ()->{                System.out.println("原始CompletableFuture办法工作");                return "捡田螺的小男孩";            }    );    CompletableFuture thenAcceptFuture = orgFuture.thenAccept((a) -> {        if ("捡田螺的小男孩".equals(a)) {            System.out.println("关注了");        }        System.out.println("先思考思考");    });    System.out.println(thenAcceptFuture.get());}

}

3. thenApply/thenApplyAsync
   CompletableFuture的thenApply办法示意，第一个工作执行实现后，执行第二个回调办法工作，会将该工作的执行后果，作为入参，传递到回调办法中，并且回调办法是有返回值的。

public class FutureThenApplyTest {

public static void main(String[] args) throws ExecutionException, InterruptedException {    CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(            ()->{                System.out.println("原始CompletableFuture办法工作");                return "捡田螺的小男孩";            }    );    CompletableFuture<String> thenApplyFuture = orgFuture.thenApply((a) -> {        if ("捡田螺的小男孩".equals(a)) {            return "关注了";        }        return "先思考思考";    });    System.out.println(thenApplyFuture.get());}

}
//输入
原始CompletableFuture办法工作
关注了

4. exceptionally
   CompletableFuture的exceptionally办法示意，某个工作执行异样时，执行的回调办法;并且有抛出异样作为参数，传递到回调办法。

public class FutureExceptionTest {

public static void main(String[] args) throws ExecutionException, InterruptedException {    CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(            ()->{                System.out.println("以后线程名称：" + Thread.currentThread().getName());                throw new RuntimeException();            }    );    CompletableFuture<String> exceptionFuture = orgFuture.exceptionally((e) -> {        e.printStackTrace();        return "你的程序异样啦";    });    System.out.println(exceptionFuture.get());}

}
//输入
以后线程名称：ForkJoinPool.commonPool-worker-1
java.util.concurrent.CompletionException: java.lang.RuntimeException
at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592)
at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
Caused by: java.lang.RuntimeException
at cn.eovie.future.FutureWhenTest.lambda$main$0(FutureWhenTest.java:13)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
... 5 more
你的程序异样啦

5. whenComplete办法
   CompletableFuture的whenComplete办法示意，某个工作执行实现后，执行的回调办法，无返回值；并且whenComplete办法返回的CompletableFuture的result是上个工作的后果。

public class FutureWhenTest {

public static void main(String[] args) throws ExecutionException, InterruptedException {    CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(            ()->{                System.out.println("以后线程名称：" + Thread.currentThread().getName());                try {                    Thread.sleep(2000L);                } catch (InterruptedException e) {                    e.printStackTrace();                }                return "捡田螺的小男孩";            }    );    CompletableFuture<String> rstFuture = orgFuture.whenComplete((a, throwable) -> {        System.out.println("以后线程名称：" + Thread.currentThread().getName());        System.out.println("上个工作执行完啦，还把" + a + "传过来");        if ("捡田螺的小男孩".equals(a)) {            System.out.println("666");        }        System.out.println("233333");    });    System.out.println(rstFuture.get());}

}
//输入
以后线程名称：ForkJoinPool.commonPool-worker-1
以后线程名称：ForkJoinPool.commonPool-worker-1
上个工作执行完啦，还把捡田螺的小男孩传过来
666
233333

6. handle办法
   CompletableFuture的handle办法示意，某个工作执行实现后，执行回调办法，并且是有返回值的;并且handle办法返回的CompletableFuture的result是回调办法执行的后果。

public class FutureHandlerTest {

public static void main(String[] args) throws ExecutionException, InterruptedException {    CompletableFuture<String> orgFuture = CompletableFuture.supplyAsync(            ()->{                System.out.println("以后线程名称：" + Thread.currentThread().getName());                try {                    Thread.sleep(2000L);                } catch (InterruptedException e) {                    e.printStackTrace();                }                  );    CompletableFuture<String> rstFuture = orgFuture.handle((a, throwable) -> {        System.out.println("上个工作执行完啦，还把" + a + "传过来");               }        System.out.println("233333");        return null;    });    System.out.println(rstFuture.get());}

}

以后线程名称：ForkJoinPool.commonPool-worker-1
上个工作执行完啦，还把捡田螺的小男孩传过来
666
关注了
多个工作组合解决
ef1d149746dc30593932c4e25ff26e04.jpeg

AND组合关系
47611ce13f4e4ec9b86377f869ff45af.jpeg

thenCombine / thenAcceptBoth / runAfterBoth都示意：将两个CompletableFuture组合起来，只有这两个都失常执行完了，才会执行某个工作。

区别在于：

thenCombine：会将两个工作的执行后果作为办法入参，传递到指定办法中，且有返回值
thenAcceptBoth: 会将两个工作的执行后果作为办法入参，传递到指定办法中，且无返回值
runAfterBoth 不会把执行后果当做办法入参，且没有返回值。
public class ThenCombineTest {

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {    CompletableFuture<String> first = CompletableFuture.completedFuture("第一个异步工作");    ExecutorService executor = Executors.newFixedThreadPool(10);    CompletableFuture<String> future = CompletableFuture            //第二个异步工作            .supplyAsync(() -> "第二个异步工作", executor)            // (w, s) -> System.out.println(s) 是第三个工作            .thenCombineAsync(first, (s, w) -> {                System.out.println(w);                System.out.println(s);                return "两个异步工作的组合";            }, executor);    System.out.println(future.join());    executor.shutdown();}

}
//输入
第一个异步工作
第二个异步工作
两个异步工作的组合
OR 组合的关系
3a8b9d26a3142a217cb5f53255d06502.jpeg

applyToEither / acceptEither / runAfterEither 都示意：将两个CompletableFuture组合起来，只有其中一个执行完了,就会执行某个工作。

区别在于：

applyToEither：会将曾经执行实现的工作，作为办法入参，传递到指定办法中，且有返回值
acceptEither: 会将曾经执行实现的工作，作为办法入参，传递到指定办法中，且无返回值
runAfterEither：不会把执行后果当做办法入参，且没有返回值。
public class AcceptEitherTest {

public static void main(String[] args) {    //第一个异步工作，休眠2秒，保障它执行晚点    CompletableFuture<String> first = CompletableFuture.supplyAsync(()->{        try{            Thread.sleep(2000L);            System.out.println("执行完第一个异步工作");}            catch (Exception e){                return "第一个工作异样";            }        return "第一个异步工作";    });    ExecutorService executor = Executors.newSingleThreadExecutor();    CompletableFuture<Void> future = CompletableFuture            //第二个异步工作            .supplyAsync(() -> {                        System.out.println("执行完第二个工作");                        return "第二个工作";}            , executor)            //第三个工作            .acceptEitherAsync(first, System.out::println, executor);    executor.shutdown();}

}
//输入
执行完第二个工作
第二个工作
AllOf
所有工作都执行实现后，才执行 allOf返回的CompletableFuture。如果任意一个工作异样，allOf的CompletableFuture，执行get办法，会抛出异样

public class allOfFutureTest {

public static void main(String[] args) throws ExecutionException, InterruptedException {    CompletableFuture<Void> a = CompletableFuture.runAsync(()->{        System.out.println("我执行完了");    });    CompletableFuture<Void> b = CompletableFuture.runAsync(() -> {        System.out.println("我也执行完了");    });    CompletableFuture<Void> allOfFuture = CompletableFuture.allOf(a, b).whenComplete((m,k)->{        System.out.println("finish");    });}

}
//输入
我执行完了
我也执行完了
finish
AnyOf
任意一个工作执行完，就执行anyOf返回的CompletableFuture。如果执行的工作异样，anyOf的CompletableFuture，执行get办法，会抛出异样

public class AnyOfFutureTest {

public static void main(String[] args) throws ExecutionException, InterruptedException {    CompletableFuture<Void> a = CompletableFuture.runAsync(()->{        try {            Thread.sleep(3000L);        } catch (InterruptedException e) {            e.printStackTrace();        }        System.out.println("我执行完了");    });    CompletableFuture<Void> b = CompletableFuture.runAsync(() -> {        System.out.println("我也执行完了");    });    CompletableFuture<Object> anyOfFuture = CompletableFuture.anyOf(a, b).whenComplete((m,k)->{        System.out.println("finish");

// return "捡田螺的小男孩";

    });    anyOfFuture.join();}

}
//输入
我也执行完了
finish
thenCompose
thenCompose办法会在某个工作执行实现后，将该工作的执行后果,作为办法入参,去执行指定的办法。该办法会返回一个新的CompletableFuture实例

如果该CompletableFuture实例的result不为null，则返回一个基于该result新的CompletableFuture实例；
如果该CompletableFuture实例为null，而后就执行这个新工作
public class ThenComposeTest {

public static void main(String[] args) throws ExecutionException, InterruptedException {    CompletableFuture<String> f = CompletableFuture.completedFuture("第一个工作");    //第二个异步工作    ExecutorService executor = Executors.newSingleThreadExecutor();    CompletableFuture<String> future = CompletableFuture            .supplyAsync(() -> "第二个工作", executor)            .thenComposeAsync(data -> {                System.out.println(data); return f; //应用第一个工作作为返回            }, executor);    System.out.println(future.join());    executor.shutdown();}

}
//输入
第二个工作
第一个工作
CompletableFuture应用有哪些留神点
CompletableFuture 使咱们的异步编程更加便当的、代码更加优雅的同时，咱们也要关注下它，应用的一些留神点。

db4f4125408776ec1224fa1ad1d52a87.jpeg

1. Future须要获取返回值，能力获取异样信息
   ExecutorService executorService = new ThreadPoolExecutor(5, 10, 5L,
   TimeUnit.SECONDS, new ArrayBlockingQueue<>(10));
   CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
   int a = 0;
   int b = 666;
   int c = b / a;
   return true;
   },executorService).thenAccept(System.out::println);

   //如果不加 get()办法这一行，看不到异样信息
   //future.get();
   Future须要获取返回值，能力获取到异样信息。如果不加 get()/join()办法，看不到异样信息。小伙伴们应用的时候，留神一下哈,思考是否加try...catch...或者应用exceptionally办法。

2. CompletableFuture的get()办法是阻塞的。
   CompletableFuture的get()办法是阻塞的，如果应用它来获取异步调用的返回值，须要增加超时工夫~

//反例
CompletableFuture.get();
//正例
CompletableFuture.get(5, TimeUnit.SECONDS);

3. 默认线程池的留神点
   CompletableFuture代码中又应用了默认的线程池，解决的线程个数是电脑CPU核数-1。在大量申请过去的时候，解决逻辑简单的话，响应会很慢。个别倡议应用自定义线程池，优化线程池配置参数。
4. 自定义线程池时，留神饱和策略
   CompletableFuture的get()办法是阻塞的，咱们个别倡议应用future.get(3, TimeUnit.SECONDS)。并且个别倡议应用自定义线程池。

然而如果线程池回绝策略是DiscardPolicy或者DiscardOldestPolicy，当线程池饱和时，会间接抛弃工作，不会摈弃异样。因而倡议，CompletableFuture线程池策略最好应用AbortPolicy，而后耗时的异步线程，做好线程池隔离哈。