CompletableFuture

CmyShare2022/1/1大约 19 分钟

背景

因为CompletableFuture实现了Future接口，我们先来回顾Future吧。Future是Java5新加的一个接口，它提供了一种异步并行计算的功能。如果主线程需要执行一个很耗时的计算任务，我们就可以通过future把这个任务放到异步线程中执行。主线程继续处理其他任务，处理完成后，再通过Future获取计算结果。来看个简单例子吧，假设我们有两个任务服务，一个查询用户基本信息，一个是查询用户勋章信息。如下，

public class UserInfoService {

    public UserInfo getUserInfo(Long userId) throws InterruptedException {
        Thread.sleep(300);//模拟调用耗时
        return new UserInfo("666", "Cmyshare", 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");
    }
}

如果我们不使用Future进行并行异步调用，而是在主线程串行进行的话，耗时大约为300+500+300 = 1100 ms。可以发现，future+线程池异步配合，提高了程序的执行效率。但是Future对于结果的获取，不是很友好，只能通过阻塞或者轮询的方式得到任务的结果。

Future.get() 就是阻塞调用，在线程获取结果之前get方法会一直阻塞。
Future提供了一个isDone方法，可以在程序中轮询这个方法查询执行结果。

阻塞的方式和异步编程的设计理念相违背，而轮询的方式会耗费无谓的CPU资源。因此，JDK8设计出CompletableFuture。CompletableFuture提供了一种观察者模式类似的机制，可以让任务执行完成后通知监听的一方。

简介

CompletableFuture是jdk8的新特性。CompletableFuture实现了CompletionStage接口和Future接口，前者是对后者的一个扩展，增加了异步会点、流式处理、多个Future组合处理的能力，使Java在处理多任务的协同工作时更加顺畅便利。CompletableFuture提供了几十种方法，辅助我们的异步任务场景。这些方法包括创建异步任务、任务异步回调、多个任务组合处理等方面。

如何使用

创建异步任务

CompletableFuture创建异步任务，一般有supplyAsync和runAsync两个方法。

supplyAsync执行CompletableFuture任务，支持返回值
runAsync执行CompletableFuture任务，没有返回值。

//使用默认内置线程池ForkJoinPool.commonPool()，根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
//自定义线程，根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

//使用默认内置线程池ForkJoinPool.commonPool()，根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
//自定义线程，根据supplier构建执行任务
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, 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,关注公众号:Cmyshare"), executor);
        //supplyAsync的使用
        CompletableFuture<String> supplyFuture = CompletableFuture.supplyAsync(() -> {
                    System.out.print("supply,关注公众号:Cmyshare");
                    return "Cmyshare"; }, executor);
        //runAsync的future没有返回值，输出null
        System.out.println(runFuture.join());
        //supplyAsync的future，有返回值
        System.out.println(supplyFuture.join());
        executor.shutdown(); // 线程池需要关闭
    }
}
//输出
run,关注公众号:Cmyshare
null
supply,关注公众号:CmyshareCmyshare

任务异步回调

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 "Cmyshare";
                }
        );

        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 "Cmyshare";
                }
        );

        CompletableFuture thenAcceptFuture = orgFuture.thenAccept((a) -> {
            if ("Cmyshare".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 "Cmyshare";
                }
        );

        CompletableFuture<String> thenApplyFuture = orgFuture.thenApply((a) -> {
            if ("Cmyshare".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 "Cmyshare";
                }
        );

        CompletableFuture<String> rstFuture = orgFuture.whenComplete((a, throwable) -> {
            System.out.println("当前线程名称：" + Thread.currentThread().getName());
            System.out.println("上个任务执行完啦，还把" + a + "传过来");
            if ("Cmyshare".equals(a)) {
                System.out.println("666");
            }
            System.out.println("233333");
        });

        System.out.println(rstFuture.get());
    }
}
//输出
当前线程名称：ForkJoinPool.commonPool-worker-1
当前线程名称：ForkJoinPool.commonPool-worker-1
上个任务执行完啦，还把Cmyshare传过来
666
233333
Cmyshare

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();
                    }
                    return "Cmyshare";
                }
        );

        CompletableFuture<String> rstFuture = orgFuture.handle((a, throwable) -> {

            System.out.println("上个任务执行完啦，还把" + a + "传过来");
            if ("Cmyshare".equals(a)) {
                System.out.println("666");
                return "关注了";
            }
            System.out.println("233333");
            return null;
        });

        System.out.println(rstFuture.get());
    }
}
//输出
当前线程名称：ForkJoinPool.commonPool-worker-1
上个任务执行完啦，还把Cmyshare传过来
666
关注了

多个任务组合处理

AND组合关系

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 组合的关系

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 "Cmyshare";
        });
        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();

    }
}
//输出
第二个任务
第一个任务

实践Demo

import java.util.concurrent.*;

/**
 * 线程创建方式：4种方式
 * 1）、继承Thread
 * Thread01 thread = new Thread01();
 * thread.start();//启动线程
 *
 * 2）、实现Runnable接口
 * Runable01 runable01 = new Runable01();
 * new Thread(runable01).start();
 *
 * 3）、实现Callable接口 + FutureTask （可以拿到返回结果，可以处理异常）
 * FutureTask<Integer> futureTask = new FutureTask<>(new Callable01());
 * new Thread(futureTask).start();
 * //阻塞等待整个线程执行完成，获取返回结果
 * Integer integer = futureTask.get();
 *
 * 4）、线程池[ExecutorService]
 * 给线程池直接提交任务。
 * service.execute(new Runable01());
 * 1、创建：
 * 1）、Executors工具类
 * 2）、new ThreadPoolExecutor原生方法
 *
 * Future:可以获取到异步结果
 *
 * 区别：1、2不能得到返回值。3可以获取返回值
 * 1、2、3都不能控制资源
 * 4可以控制资源，性能稳定。
 */
public class ThreadTest {
    /**
     * Executors创建线程池
     * Executors.newCachedThreadPool() core是0，所有都可回收
     * Executors.newFixedThreadPool() 固定大小，core=max；都不可回收
     * Executors.newScheduledThreadPool() 定时任务的线程池
     * Executors.newSingleThreadExecutor() 单线程的线程池，后台从队列里面获取任务，挨个执行
     */
    public static ExecutorService executor = Executors.newFixedThreadPool(10);

    /**
     * ThreadPoolExecutor创建线程池
     *
     * @param args
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void thread(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main....start....");

        //业务代码里面，以上三种启动线程的方式都不用。【将所有的多线程异步任务都交给线程池执行】
        //当前系统中池只有一两个，每个异步任务，提交给线程池让他自己去执行就行
        //new Thread(()-> System.out.println("hello")).start();

        /**
         * ThreadPoolExecutor七大参数:
         * corePoolSize:[5] 核心线程数[一直存在除非（allowCoreThreadTimeOut）]; 线程池，创建好以后就准备就绪的线程数量，就等待来接受异步任务去执行。
         *        5个 Thread thread = new Thread();  thread.start();
         *
         * maximumPoolSize:[200] 最大线程数量;  控制资源
         *
         * keepAliveTime:存活时间。如果当前的线程数量大于core数量。
         *      释放空闲的线程（maximumPoolSize-corePoolSize）。只要线程空闲大于指定的keepAliveTime；unit:时间单位
         *
         * BlockingQueue<Runnable> workQueue:阻塞队列。如果任务有很多，就会将目前多的任务放在队列里面。
         *              只要有线程空闲，就会去队列里面取出新的任务继续执行。
         *
         * threadFactory:线程的创建工厂，默认或自定义。
         *
         * RejectedExecutionHandler handler:如果队列满了，按照我们指定的拒绝策略拒绝执行任务
         *
         * ThreadPoolExecutor工作顺序:
         * 1)、线程池创建，准备好core数量的核心线程，准备接受任务
         * 1.1、core满了，就将再进来的任务放入阻塞队列中。空闲的core就会自己去阻塞队列获取任务执行
         * 1.2、阻塞队列满了，就直接开新线程执行，最大只能开到max指定的数量
         * 1.3、max满了就用RejectedExecutionHandler拒绝任务
         * 1.4、max都执行完成，有很多空闲.在指定的时间keepAliveTime以后，释放max-core这些线程
         * new LinkedBlockingDeque<>()：链接阻塞双端队列默认是Integer的最大值。导致内存不够
         *
         * 面试题：一个线程池 core 7； max 20 ，queue：50，100并发进来怎么分配的；
         * 7个会立即得到执行，50个会进入队列，再开13个进行执行。剩下的30个就使用拒绝策略。
         * 如果不想抛弃还要执行。CallerRunsPolicy；
         *
         */
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5,
                200,
                10,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(100000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
        System.out.println("main....end....");
    }

    /**
     * 继承Thread
     */
    public static class Thread01 extends Thread {
        @Override
        public void run() {
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        }
    }

    /**
     * 实现Runnable
     */
    public static class Runable01 implements Runnable {

        @Override
        public void run() {
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        }
    }

    /**
     * 实现Callable<Integer>
     */
    public static class Callable01 implements Callable<Integer> {

        @Override
        public Integer call() throws Exception {
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
            return i;
        }
    }


    /**
     * CompletableFuture异步编程
     * @param args
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main....start....");

        /**
         * CompletableFuture-runAsync：简单创建无返回值的异步任务
         */
        //CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
        //    //任务内容
        //    System.out.println("当前线程：" + Thread.currentThread().getId());
        //    int i = 10 / 2;
        //    System.out.println("运行结果：" + i);
        //}, executor);

        /**
         * CompletableFuture-supplyAsync：方法完成后的感知whenComplete、exceptionally
         */
        //CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        //    System.out.println("当前线程：" + Thread.currentThread().getId());
        //    int i = 10 / 0;
        //    System.out.println("运行结果：" + i);
        //    return i;
        //}, executor).whenComplete((res,excption)->{
        //    //whenComplete监听，虽然能得到异常信息，但是没法修改返回数据。
        //    System.out.println("异步任务成功完成了...结果是："+res+";异常是:"+excption);
        //}).exceptionally(throwable -> {
        //    //exceptionally监听，感知异常，同时返回默认值
        //    return 10;
        //});
        ////获取异步任务返回值
        //Integer integer = future.get();
        //System.out.println("方法完成后的感知"+integer);


        /**
         * CompletableFuture-supplyAsync：方法执行完成后的处理handle
         */
        //CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        //    System.out.println("当前线程：" + Thread.currentThread().getId());
        //    int i = 10 / 4;
        //    System.out.println("运行结果：" + i);
        //    return i;
        //}, executor).handle((res, thr) -> {
        //    if (res != null) {
        //        return res * 2;
        //    }
        //    if (thr != null) {
        //        return 0;
        //    }
        //    return 0;
        //});
        ////获取异步任务返回值
        //Integer integer = future.get();
        //System.out.println("方法完成后的感知"+integer);

        /**
         * CompletableFuture线程串行化：
         * 1）、thenRun：不能获取到上一步的执行结果，无返回值
         *  .thenRunAsync(() -> {
         *             System.out.println("任务2启动了...");
         *         }, executor);
         * 2）、thenAcceptAsync;能接受上一步结果，但是无返回值
         * 3）、thenApplyAsync：;能接受上一步结果，有返回值
         */
        //CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        //    System.out.println("当前线程：" + Thread.currentThread().getId());
        //    int i = 10 / 4;
        //    System.out.println("运行结果：" + i);
        //    return i;
        //}, executor).thenApplyAsync(res -> {
        //    System.out.println("任务2启动了..." + res);
        //    return "Hello " + res;
        //}, executor);
        //System.out.println("输出最后返回值"+future.get());

        /**
         * 两任务组合 - 都要完成
         * thenCombine：组合两个future，获取两个future的返回结果，并返回当前任务的返回值
         * thenAcceptBoth：组合两个future，获取两个future任务的返回结果，然后处理任务，没有返回值。
         * runAfterBoth：组合两个future，不需要获取两个future的结果，只需两个future处理完任务后，处理该任务。
         */
        //CompletableFuture<Object> future01 = CompletableFuture.supplyAsync(() -> {
        //    System.out.println("任务1线程：" + Thread.currentThread().getId());
        //    int i = 10 / 4;
        //    System.out.println("任务1结束：" );
        //    return i;
        //}, executor);
        //
        //CompletableFuture<Object> future02 = CompletableFuture.supplyAsync(() -> {
        //    System.out.println("任务2线程：" + Thread.currentThread().getId());
        //    try {
        //        //任务2睡眠3000毫秒
        //        Thread.sleep(3000);
        //        System.out.println("任务2结束：" );
        //    } catch (InterruptedException e) {
        //        e.printStackTrace();
        //    }
        //    return "Hello";
        //}, executor);
        //
        ////任务组合runAfterBothAsync
        //future01.runAfterBothAsync(future02,()->{
        //    System.out.println("任务3开始...");
        //},executor);
        //
        ////任务组合thenAcceptBothAsync
        //future01.thenAcceptBothAsync(future02,(f1,f2)->{
        //    System.out.println("任务3开始...之前的结果："+f1+"--》"+f2);
        //},executor);
        //
        ////任务组合thenCombineAsync
        //CompletableFuture<String> future = future01.thenCombineAsync(future02, (f1, f2) -> {
        //    return f1 + "：" + f2 + " -> Haha";
        //}, executor);

        /**
         * 两个任务，只要有一个完成，我们就执行任务3
         * runAfterEitherAsync：不感知结果，自己没有返回值
         * acceptEitherAsync：感知结果，自己没有返回值
         * applyToEitherAsync：感知结果，自己有返回值
         */
        //future01.runAfterEitherAsync(future02,()->{
        //    System.out.println("任务3开始...之前的结果：");
        //},executor);
        //future01.acceptEitherAsync(future02,(res)->{
        //    System.out.println("任务3开始...之前的结果："+res);
        //},executor);
        //CompletableFuture<String> future = future01.applyToEitherAsync(future02, res -> {
        //    System.out.println("任务3开始...之前的结果：" + res);
        //    return res.toString() + "->哈哈";
        //}, executor);


        /**
         * 多任务组合：
         * allOf： 等待所有任务完成
         * anyOf： 只要有一个任务完成
         */
        CompletableFuture<String> futureImg = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品的图片信息");
            return "hello.jpg";
        }, executor);

        CompletableFuture<String> futureAttr = CompletableFuture.supplyAsync(() -> {
            System.out.println("查询商品的属性");
            return "黑色+256G";
        }, executor);

        CompletableFuture<String> futureDesc = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
                System.out.println("查询商品介绍");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "华为";
        }, executor);

        //CompletableFuture<Void> allOf = CompletableFuture.allOf(futureImg, futureAttr, futureDesc);
        CompletableFuture<Object> anyOf = CompletableFuture.anyOf(futureImg, futureAttr, futureDesc);
        anyOf.get();//阻塞等待所有结果完成，也可以用join

        //System.out.println("main....end...."+futureImg.get()+"=>"+futureAttr.get()+"=>"+futureDesc.get());
        System.out.println("main....end...." + anyOf.get());
    }

}

使用有哪些注意点

CompletableFuture 使我们的异步编程更加便利的、代码更加优雅的同时，我们也要关注下它，使用的一些注意点。

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，然后耗时的异步线程，做好线程池隔离哈。

1、使用whenComplete方法
whenComplete方法可以在任务完成时触发回调函数，无论是正常完成还是发生异常。通过在whenComplete方法中处理异常，我们可以确保异常得到正确的传播和处理。
whenComplete获取异常信息: 异常是存在于异步当中的, 不能被主线程捕获

2、使用exceptionally方法处理异常
exceptionally方法可以用于处理异常，并返回一个默认值或执行其他操作。在使用exceptionally方法时，我们需要确保正确地处理异常并重新抛出，以便异常能够传播到后续阶段。
exceptionally获取异常信息: 异常是存在于异步当中的, 不能被主线程捕获。

3、使用handle方法处理异常
handle方法可以用于处理异常，并根据需要返回一个新的结果。与exceptionally方法不同的是，handle方法可以处理正常的返回结果和异常，并返回一个新的结果。
handle获取异常信息: 异常是存在于异步当中的, 不能被主线程捕获

4. 使用CompletableFuture.allOf组合多个CompletableFutur
当需要处理多个CompletableFuture时，我们可以使用CompletableFuture.allOf方法来组合它们，并在所有任务完成后进行处理。通过使用whenComplete或exceptionally方法，我们可以处理所有任务的异常，并确保异常正确地传播和处理。

5、异步调用CompletableFuture.join方法
join方法获取异常信息: 将异步线程中发生的异常信息抛到主线程, 这样异常可被主线程捕获

6、异步调用CompletableFuture.get()
 get方法获取异常信息: 将异步线程中发生的异常信息抛到主线程, 这样异常可被主线程捕获。异步方法中get()是阻塞的，在使用时要设置超时时间。

总结

在使用异步CompletableFuture时，无论是否有返回值都要调用get()/join()方法，避免程序执行报错了，仍然返回成功。如果在程序报错时需要对上一个异步任务结果做其他操作，可以调用whenComplete()、handle()处理，如果只是对异常做处理，不涉及对上一个异步任务结果的情况，调用exceptionally()处理。

参考链接

官方文档：https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html https://juejin.cn/post/6970558076642394142#heading-20 https://blog.csdn.net/zsx_xiaoxin/article/details/123898171

异常处理 https://juejin.cn/post/7324933059819962394#heading-6 https://blog.csdn.net/m0_73311735/article/details/131456184 https://www.cnblogs.com/song27/p/15146248.html https://blog.csdn.net/winterking3/article/details/116489984

贡献者

cmyshare