乐趣区

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

jiezi

前言
最近刚好应用 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());
}

}

  1. 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 办法工作
关注了

  1. 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
你的程序异样啦

  1. 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

  1. 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);

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

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

退出移动版