并发编程(九)-CompletableFuture原理与实践

测试 thenCombine：

• 第一种写法

@Slf4j(topic = "c.TestThenCombine")
public class TestThenCombine {
    public static void main(String[] args) {
        log.debug("开始");
        CompletableFuture A = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("A 完成");
            return 10;
        });

        CompletableFuture B = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("B 完成");
            return 20;
        });

        CompletableFuture result = A.thenCombine(B, (x, y) -> x + y);
        log.debug("结果：" + result.join());
    }
}

16:47:50.931 c.TestThenCombine [main] - 开始
16:47:51.992 c.TestThenCombine [ForkJoinPool.commonPool-worker-9] - A 完成
16:47:53.007 c.TestThenCombine [ForkJoinPool.commonPool-worker-2] - B 完成
16:47:53.007 c.TestThenCombine [main] - 结果：30

• 第二种写法

@Slf4j(topic = "c.TestThenCombine")
public class TestThenCombine {
    public static void main(String[] args) {
        log.debug("开始");
        CompletableFuture result = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("A 完成");
            return 10;
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.debug("B 完成");
            return 20;
        }), (x, y) -> x + y);

        log.debug("结果：" + result.join());
    }
}

16:53:50.817 c.TestThenCombine [main] - 开始
16:53:51.894 c.TestThenCombine [ForkJoinPool.commonPool-worker-9] - A 完成
16:53:52.886 c.TestThenCombine [ForkJoinPool.commonPool-worker-2] - B 完成
16:53:52.886 c.TestThenCombine [main] - 结果：30

• 当 任务数 等于 CompletableFuture 线程池的最大线程数（CPU核心数 - 1） 时，CompletableFuture 性能最佳
  例如：有4个核心，则 CompletableFuture 线程池的最大线程数是3，此时任务数为3（假设每个任务耗时1s），CompletableFuture 性能最 佳（CompletableFuture 做完3个任务耗时也就1s左右）；
  如果任务数为4-6，则 CompletableFuture 做完4-6个任务耗时就会是2s左右；
  如果任务数为7-9，则 CompletableFuture 做完7-9个任务耗时就会是3s左右；
  以此类推
• 所以当已知任务数量时，可以创建自定义线程池，手动设置线程池最大线程数
• 如果需要根据任务数量动态设置线程池最大线程数时，每执行完一批任务后，需要关闭线程池，等下一次新任务来时再根据任务数重新创建线程池
  例如：第一次来了100个任务，则创建10个线程的线程池，执行完100个任务后关闭线程池；第二次来了200个任务时，再重新创建有20个线程的线程池；

测试 任务数 等于 CompletableFuture 线程池的最大线程数：

创建任务类：

@Slf4j(topic = "c.Task")
class Task {
    // 任务名
    private String name;
    // 做该任务耗时 (秒)
    private Integer productionTime;

    public Task(String name, Integer productionTime) {
        this.name = name;
        this.productionTime = productionTime;
    }

    // 做任务
    public void doTask() {
        try {
            TimeUnit.SECONDS.sleep(this.productionTime);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.debug(this.name + " 已做完");
    }
}

第一种写法：

@Slf4j(topic = "c.TestCompletableFuture_doTask")
public class TestCompletableFuture_doTask {
    public static void main(String[] args) {
		log.debug("该二手台式机可用处理器数量：{}", Runtime.getRuntime().availableProcessors());
		log.debug("当前线程池中的最大线程数：{}", ForkJoinPool.getCommonPoolParallelism());
        log.debug("开始");
        long startTime = System.currentTimeMillis();
        // 创建任务
        List tasks = new ArrayList<>();
        for (int i = 1; i <= 3; i++) {
            Task task = new Task("任务" + i, 1);
            tasks.add(task);
        }
        // 做任务
        List completableFutureList = new ArrayList<>();
        for (Task task : tasks) {
            CompletableFuture cf = CompletableFuture.runAsync(() -> task.doTask());
            // 将所有任务加到 CompletableFuture 集合中
            completableFutureList.add(cf);
        }
        // 等待 CompletableFuture 集合中的所有任务执行完毕
        CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[completableFutureList.size()])).join();

        log.debug("做完所有任务的耗时：{}", (System.currentTimeMillis() - startTime));

    }
}

00:50:41.392 c.TestCompletableFuture_doTask [main] - 该二手台式机可用处理器数量：4
00:50:41.392 c.TestCompletableFuture_doTask [main] - 当前线程池中的最大线程数：3
00:50:41.397 c.TestCompletableFuture_doTask [main] - 开始
00:50:42.467 c.Task [ForkJoinPool.commonPool-worker-3] - 任务3 已做完
00:50:42.467 c.Task [ForkJoinPool.commonPool-worker-2] - 任务2 已做完
00:50:42.467 c.Task [ForkJoinPool.commonPool-worker-1] - 任务1 已做完
00:50:42.467 c.TestCompletableFuture_doTask [main] - 做完所有任务的耗时：1069

第二种写法（用Stream简化代码，结果还是一样，主要是简化代码）：

@Slf4j(topic = "c.TestCompletableFuture_doTask")
public class TestCompletableFuture_doTask {
    public static void main(String[] args) {
		log.debug("该二手台式机可用处理器数量：{}", Runtime.getRuntime().availableProcessors());
        log.debug("当前线程池中的最大线程数：{}", ForkJoinPool.getCommonPoolParallelism());
        log.debug("开始");
        long startTime = System.currentTimeMillis();
        // 将创建的任务加到 CompletableFuture 数组中 并执行
        CompletableFuture[] tasks= IntStream.rangeClosed(1, 3)
                .mapToObj(i -> new Task("任务" + i, 1))
                .map(task -> CompletableFuture.runAsync(task::doTask))
                .toArray(CompletableFuture[]::new);
		// 等待 CompletableFuture 数组中的所有任务执行完
        CompletableFuture.allOf(tasks).join();

        log.debug("做完所有任务的耗时：{}", (System.currentTimeMillis() - startTime));
    }
}

00:51:13.655 c.TestCompletableFuture_doTask [main] - 该二手台式机可用处理器数量：4
00:51:13.655 c.TestCompletableFuture_doTask [main] - 当前线程池中的最大线程数：3
00:51:13.667 c.TestCompletableFuture_doTask [main] - 开始
00:51:14.896 c.Task [ForkJoinPool.commonPool-worker-2] - 任务2 已做完
00:51:14.896 c.Task [ForkJoinPool.commonPool-worker-1] - 任务1 已做完
00:51:14.896 c.Task [ForkJoinPool.commonPool-worker-3] - 任务3 已做完
00:51:14.896 c.TestCompletableFuture_doTask [main] - 做完所有任务的耗时：1229

测试 任务数 超过 CompletableFuture 线程池的最大线程数：

@Slf4j(topic = "c.TestCompletableFuture_doTask")
public class TestCompletableFuture_doTask {
    public static void main(String[] args) {
		log.debug("该二手台式机可用处理器数量：{}", Runtime.getRuntime().availableProcessors());
        log.debug("当前线程池中的最大线程数：{}", ForkJoinPool.getCommonPoolParallelism());
        log.debug("开始");
        long startTime = System.currentTimeMillis();
        // 将创建的任务加到 CompletableFuture 数组中 并执行
        CompletableFuture[] tasks= IntStream.rangeClosed(1, 3)
                .mapToObj(i -> new Task("任务" + i, 1))
                .map(task -> CompletableFuture.runAsync(task::doTask))
                .toArray(CompletableFuture[]::new);
		// 等待 CompletableFuture 数组中的所有任务执行完
        CompletableFuture.allOf(tasks).join();

        log.debug("做完所有任务的耗时：{}", (System.currentTimeMillis() - startTime));
    }
}

01:06:31.516 c.TestCompletableFuture_doTask [main] - 该二手台式机可用处理器数量：4
01:06:31.516 c.TestCompletableFuture_doTask [main] - 当前线程池中的最大线程数：3
01:06:31.527 c.TestCompletableFuture_doTask [main] - 开始
01:06:32.636 c.Task [ForkJoinPool.commonPool-worker-2] - 任务2 已做完
01:06:32.636 c.Task [ForkJoinPool.commonPool-worker-1] - 任务1 已做完
01:06:32.637 c.Task [ForkJoinPool.commonPool-worker-3] - 任务3 已做完
01:06:33.637 c.Task [ForkJoinPool.commonPool-worker-2] - 任务4 已做完
01:06:33.637 c.TestCompletableFuture_doTask [main] - 做完所有任务的耗时：2110

CompletableFuture 配合 自定义线程池 使用

@Slf4j(topic = "c.TestCompletableFuture_doTask")
public class TestCompletableFuture_doTask {
    public static void main(String[] args) {
		ExecutorService pool = Executors.newCachedThreadPool();
        log.debug("开始");
        long startTime = System.currentTimeMillis();
        // 将创建的任务加到 CompletableFuture 数组中 并执行
        CompletableFuture[] tasks= IntStream.rangeClosed(1, 3)
                .mapToObj(i -> new Task("任务" + i, 1))
                .map(task -> CompletableFuture.runAsync(task::doTask, pool)) // 传入自定义线程池
                .toArray(CompletableFuture[]::new);
		// 等待 CompletableFuture 数组中的所有任务执行完
        CompletableFuture.allOf(tasks).join();

        log.debug("做完所有任务的耗时：{}", (System.currentTimeMillis() - startTime));
        pool.shutdown();
    }
}

01:32:25.794 c.TestCompletableFuture_doTask [main] - 开始
01:32:26.871 c.Task [pool-1-thread-10] - 任务10 已做完
01:32:26.872 c.Task [pool-1-thread-9] - 任务9 已做完
01:32:26.872 c.Task [pool-1-thread-8] - 任务8 已做完
01:32:26.872 c.Task [pool-1-thread-7] - 任务7 已做完
01:32:26.872 c.Task [pool-1-thread-6] - 任务6 已做完
01:32:26.873 c.Task [pool-1-thread-5] - 任务5 已做完
01:32:26.873 c.Task [pool-1-thread-4] - 任务4 已做完
01:32:26.873 c.Task [pool-1-thread-3] - 任务3 已做完
01:32:26.874 c.Task [pool-1-thread-2] - 任务2 已做完
01:32:26.874 c.Task [pool-1-thread-1] - 任务1 已做完
01:32:26.874 c.TestCompletableFuture_doTask [main] - 做完所有任务的耗时：1080

可以看到就算做10个任务，耗时也是1s左右，但不是任何数量的任务都只耗时1s
一台计算机，一瞬间处理的最大任务数量跟CPU的处理能力有关，也跟内存、硬盘、网卡等设备有关
具体跟什么有关，取决于任务本身所需要的资源
由于测试案例中的任务使用sleep模拟耗时的，但是sleep这个操作不太占CPU资源，
所以当传入自定义线程池后，任务数超过了CPU核心数后，耗时也不多

随着业务量的持续上升，各系统服务面临的压力也越来越大，业务对系统吞吐量的要求也越来越高。
例如：有一个对外暴露的接口服务，所有数据都需要通过该接口进行调度，最后再将数据进行聚合返回；
内部实现其实就是该服务对内调度各个下游服务，然后再获取数据进行聚合，这种方式具有鲜明的I/O密集型（I/O Bound）特点。
如果通过同步方式进行加载数据，则下游服务越多，耗时越长，因此考虑通过并行方式加载。

并行从下游获取数据，从IO模型上来讲分为同步模型和异步模型。

从各个服务获取数据最常见的是同步调用，如下图所示：

在同步调用的场景下，接口耗时长、性能差，接口响应时长T > T1+T2+T3+……+Tn，这时为了缩短接口的响应时间，一般会使用线程池的方式并行获取数据，如下图所示：

这种方式由于以下两个原因，导致资源利用率比较低：

• CPU资源大量浪费在阻塞等待上，导致CPU资源利用率低。在Java 8之前，一般会通过回调的方式来减少阻塞，但是大量使用回调，又引发臭名昭著的回调地狱问题，导致代码可读性和可维护性大大降低。
• 为了增加并发度，会引入更多额外的线程池，随着CPU调度线程数的增加，会导致更严重的资源争用，宝贵的CPU资源被损耗在上下文切换上，而且线程本身也会占用系统资源，且不能无限增加。
  同步模型下，会导致硬件资源无法充分利用，系统吞吐量容易达到瓶颈。

主要通过以下两种方式来减少线程池的调度开销和阻塞时间：

• 通过RPC NIO异步调用的方式可以降低线程数，从而降低调度（上下文切换）开销，如Dubbo的异步调用。
  通过引入CompletableFuture（下文简称CF）对业务流程进行编排，降低依赖之间的阻塞。本文主要讲述CompletableFuture的使用和原理。

Future、CompletableFuture、RxJava、Reactor 对比如下：

• 可组合：可以将多个依赖操作通过不同的方式进行编排，例如CompletableFuture提供thenCompose、thenCombine等各种then开头的方法，这些方法就是对“可组合”特性的支持。
• 操作融合：将数据流中使用的多个操作符以某种方式结合起来，进而降低开销（时间、内存）。
• 延迟执行：操作不会立即执行，当收到明确指示时操作才会触发。例如：Reactor只有当有订阅者订阅时，才会触发操作。
• 回压：某些异步阶段的处理速度跟不上，直接失败会导致大量数据的丢失，对业务来说是不能接受的，这时需要反馈上游生产者降低调用量。

RxJava与Reactor显然更加强大，它们提供了更多的函数调用方式，支持更多特性，但同时也带来了更大的学习成本。而我们本次整合最需要的特性就是“异步”、“可组合”，综合考虑后，我们选择了学习成本相对较低的CompletableFuture。

CompletableFuture是由Java 8引入的，在Java8之前我们一般通过Future实现异步。

• Future用于表示异步计算的结果，只能通过阻塞或者轮询的方式获取结果，而且不支持设置回调方法，Java 8之前若要设置回调一般会使用guava的ListenableFuture，回调的引入又会导致臭名昭著的回调地狱。
• CompletableFuture对Future进行了扩展，可以通过设置回调的方式处理计算结果，同时也支持组合操作，支持进一步的编排，同时一定程度解决了回调地狱的问题。

下面将举例来说明，我们通过ListenableFuture、CompletableFuture来实现异步的差异。假设有三个操作step1、step2、step3存在依赖关系，其中step3的执行依赖step1和step2的结果。

Future(ListenableFuture)的实现（回调地狱）如下：

ExecutorService executor = Executors.newFixedThreadPool(5);
ListeningExecutorService guavaExecutor = MoreExecutors.listeningDecorator(executor);
ListenableFuture future1 = guavaExecutor.submit(() -> {
    //step 1
    System.out.println("执行step 1");
    return "step1 result";
});
ListenableFuture future2 = guavaExecutor.submit(() -> {
    //step 2
    System.out.println("执行step 2");
    return "step2 result";
});
ListenableFuture> future1And2 = Futures.allAsList(future1, future2);
Futures.addCallback(future1And2, new FutureCallback>() {
    @Override
    public void onSuccess(List result) {
        System.out.println(result);
        ListenableFuture future3 = guavaExecutor.submit(() -> {
            System.out.println("执行step 3");
            return "step3 result";
        });
        Futures.addCallback(future3, new FutureCallback() {
            @Override
            public void onSuccess(String result) {
                System.out.println(result);
            }        
            @Override
            public void onFailure(Throwable t) {
            }
        }, guavaExecutor);
    }

    @Override
    public void onFailure(Throwable t) {
    }}, guavaExecutor);

CompletableFuture的实现如下：

ExecutorService executor = Executors.newFixedThreadPool(5);
CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
    System.out.println("执行step 1");
    return "step1 result";
}, executor);
CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
    System.out.println("执行step 2");
    return "step2 result";
});
cf1.thenCombine(cf2, (result1, result2) -> {
    System.out.println(result1 + " , " + result2);
    System.out.println("执行step 3");
    return "step3 result";
}).thenAccept(result3 -> System.out.println(result3));

显然，CompletableFuture的实现更为简洁，可读性更好。

下面我们通过一个例子来讲解CompletableFuture如何使用，使用CompletableFuture也是构建依赖树的过程。一个CompletableFuture的完成会触发另外一系列依赖它的CompletableFuture的执行：

如上图所示，这里描绘的是一个业务接口的流程，其中包括CF1CF2CF3CF4CF5共5个步骤，并描绘了这些步骤之间的依赖关系，每个步骤可以是一次RPC调用、一次数据库操作或者是一次本地方法调用等，在使用CompletableFuture进行异步化编程时，图中的每个步骤都会产生一个CompletableFuture对象，最终结果也会用一个CompletableFuture来进行表示。

根据CompletableFuture依赖数量，可以分为以下几类：零依赖、一元依赖、二元依赖和多元依赖。

我们先看下如何不依赖其他CompletableFuture来创建新的CompletableFuture：

如上图红色链路所示，接口接收到请求后，首先发起两个异步调用CF1、CF2，主要有三种方式：

ExecutorService executor = Executors.newFixedThreadPool(5);
//1、使用runAsync或supplyAsync发起异步调用
CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  return "result1";
}, executor);
//2、CompletableFuture.completedFuture()直接创建一个已完成状态的CompletableFuture
CompletableFuture cf2 = CompletableFuture.completedFuture("result2");
//3、先初始化一个未完成的CompletableFuture，然后通过complete()、completeExceptionally()，完成该CompletableFuture
CompletableFuture cf = new CompletableFuture<>();
cf.complete("success");

第三种方式的一个典型使用场景，就是将回调方法转为CompletableFuture，然后再依赖CompletableFure的能力进行调用编排，示例如下：

@FunctionalInterface
public interface ThriftAsyncCall {
    void invoke() throws TException;
}
 
  public static  CompletableFuture toCompletableFuture(final OctoThriftCallback callback , ThriftAsyncCall thriftCall) {
   //新建一个未完成的CompletableFuture
   CompletableFuture resultFuture = new CompletableFuture<>();
   //监听回调的完成，并且与CompletableFuture同步状态
   callback.addObserver(new OctoObserver() {
       @Override
       public void onSuccess(T t) {
           resultFuture.complete(t);
       }
       @Override
       public void onFailure(Throwable throwable) {
           resultFuture.completeExceptionally(throwable);
       }
   });
   if (thriftCall != null) {
       try {
           thriftCall.invoke();
       } catch (TException e) {
           resultFuture.completeExceptionally(e);
       }
   }
   return resultFuture;
  }

如上图红色链路所示，CF3，CF5分别依赖于CF1和CF2，这种对于单个CompletableFuture的依赖可以通过thenApply、thenAccept、thenCompose等方法来实现，代码如下所示：

CompletableFuture cf3 = cf1.thenApply(result1 -> {
  //result1为CF1的结果
  //......
  return "result3";
});
CompletableFuture cf5 = cf2.thenApply(result2 -> {
  //result2为CF2的结果
  //......
  return "result5";
});

如上图红色链路所示，CF4同时依赖于两个CF1和CF2，这种二元依赖可以通过thenCombine等回调来实现，如下代码所示：

CompletableFuture cf4 = cf1.thenCombine(cf2, (result1, result2) -> {
  //result1和result2分别为cf1和cf2的结果
  return "result4";
});

如上图红色链路所示，整个流程的结束依赖于三个步骤CF3、CF4、CF5，这种多元依赖可以通过allOf或anyOf方法来实现，区别是当需要多个依赖全部完成时使用allOf，当多个依赖中的任意一个完成即可时使用anyOf，以allOf为例，如下代码所示：

// 1.当需要多个依赖全部完成时使用allOf
CompletableFuture cf6 = CompletableFuture.allOf(cf3, cf4, cf5);
CompletableFuture result = cf6.thenApply(v -> {
  //这里的join并不会阻塞，因为传给thenApply的函数是在CF3、CF4、CF5全部完成时，才会执行 。
  result3 = cf3.join();
  result4 = cf4.join();
  result5 = cf5.join();
  //根据result3、result4、result5组装最终result;
  return "result";
});

CompletableFuture中包含两个字段：result和stack。result用于存储当前CF的结果，stack（Completion）表示当前CF完成后需要触发的依赖动作（Dependency Actions），去触发依赖它的CF的计算，依赖动作可以有多个（表示有多个依赖它的CF），以栈（Treiber stack）的形式存储，stack表示栈顶元素。

这种方式类似“观察者模式”，依赖动作（Dependency Action）都封装在一个单独Completion子类中。下面是Completion类关系结构图。CompletableFuture中的每个方法都对应了图中的一个Completion的子类，Completion本身是观察者的基类。

• UniCompletion继承了Completion，是一元依赖的基类，例如thenApply的实现类UniApply就继承自UniCompletion。
• BiCompletion继承了UniCompletion，是二元依赖的基类，同时也是多元依赖的基类。例如thenCombine的实现类BiRelay就继承自BiCompletion。
  

按照类似“观察者模式”的设计思想，原理分析可以从“观察者”和“被观察者”两个方面着手。由于回调种类多，但结构差异不大，所以这里单以一元依赖中的thenApply为例，不再枚举全部回调类型。如下图所示：

被观察者：

1. 每个CompletableFuture都可以被看作一个被观察者，其内部有一个Completion类型的链表成员变量stack，用来存储注册到其中的所有观察者。当被观察者执行完成后会弹栈stack属性，依次通知注册到其中的观察者。上面例子中步骤fn2就是作为观察者被封装在UniApply中。
2. 被观察者CF中的result属性，用来存储返回结果数据。这里可能是一次RPC调用的返回值，也可能是任意对象，在上面的例子中对应步骤fn1的执行结果。

观察者：
CompletableFuture支持很多回调方法，例如thenAccept、thenApply、exceptionally等，这些方法接收一个函数类型的参数f，生成一个Completion类型的对象（即观察者），并将入参函数f赋值给Completion的成员变量fn，然后检查当前CF是否已处于完成状态（即result != null），如果已完成直接触发fn，否则将观察者Completion加入到CF的观察者链stack中，再次尝试触发，如果被观察者未执行完则其执行完毕之后通知触发。

1. 观察者中的dep属性：指向其对应的CompletableFuture，在上面的例子中dep指向CF2。
2. 观察者中的src属性：指向其依赖的CompletableFuture，在上面的例子中src指向CF1。
3. 观察者Completion中的fn属性：用来存储具体的等待被回调的函数。这里需要注意的是不同的回调方法（thenAccept、thenApply、exceptionally等）接收的函数类型也不同，即fn的类型有很多种，在上面的例子中fn指向fn2。

这里仍然以thenApply为例来说明一元依赖的流程：

1. 将观察者Completion注册到CF1，此时CF1将Completion压栈。
2. 当CF1的操作运行完成时，会将结果赋值给CF1中的result属性。
3. 依次弹栈，通知观察者尝试运行。
   
   初步流程设计如上图所示，这里有几个关于注册与通知的并发问题，大家可以思考下：

Q1：在观察者注册之前，如果CF已经执行完成，并且已经发出通知，那么这时观察者由于错过了通知是不是将永远不会被触发呢 ？ A1：不会。在注册时检查依赖的CF是否已经完成。如果未完成（即result == null）则将观察者入栈，如果已完成（result != null）则直接触发观察者操作。

Q2：在”入栈“前会有”result == null“的判断，这两个操作为非原子操作，CompletableFufure的实现也没有对两个操作进行加锁，完成时间在这两个操作之间，观察者仍然得不到通知，是不是仍然无法触发？

A2：不会。入栈之后再次检查CF是否完成，如果完成则触发。

Q3：当依赖多个CF时，观察者会被压入所有依赖的CF的栈中，每个CF完成的时候都会进行，那么会不会导致一个操作被多次执行呢 ？如下图所示，即当CF1、CF2同时完成时，如何避免CF3被多次触发。

A3：CompletableFuture的实现是这样解决该问题的：观察者在执行之前会先通过CAS操作设置一个状态位，将status由0改为1。如果观察者已经执行过了，那么CAS操作将会失败，取消执行。

通过对以上3个问题的分析可以看出，CompletableFuture在处理并行问题时，全程无加锁操作，极大地提高了程序的执行效率。我们将并行问题考虑纳入之后，可以得到完善的整体流程图如下所示：

CompletableFuture支持的回调方法十分丰富，但是正如上一章节的整体流程图所述，他们的整体流程是一致的。所有回调复用同一套流程架构，不同的回调监听通过策略模式实现差异化。

我们以thenCombine为例来说明二元依赖：
thenCombine操作表示依赖两个CompletableFuture。其观察者实现类为BiApply，如上图所示，BiApply通过src和snd两个属性关联被依赖的两个CF，fn属性的类型为BiFunction。与单个依赖不同的是，在依赖的CF未完成的情况下，thenCombine会尝试将BiApply压入这两个被依赖的CF的栈中，每个被依赖的CF完成时都会尝试触发观察者BiApply，BiApply会检查两个依赖是否都完成，如果完成则开始执行。这里为了解决重复触发的问题，同样用的是上一章节提到的CAS操作，执行时会先通过CAS设置状态位，避免重复触发。

依赖多个CompletableFuture的回调方法包括allOf、anyOf，区别在于allOf观察者实现类为BiRelay，需要所有被依赖的CF完成后才会执行回调；而anyOf观察者实现类为OrRelay，任意一个被依赖的CF完成后就会触发。二者的实现方式都是将多个被依赖的CF构建成一棵平衡二叉树，执行结果层层通知，直到根节点，触发回调监听。

要合理治理线程资源，最基本的前提条件就是要在写代码时，清楚地知道每一行代码都将执行在哪个线程上。下面我们看一下CompletableFuture的执行线程情况。

CompletableFuture实现了CompletionStage接口，通过丰富的回调方法，支持各种组合操作，每种组合场景都有同步和异步两种方法。

同步方法（即不带Async后缀的方法）有两种情况：

• 如果注册时被依赖的操作已经执行完成，则直接由当前线程执行。
• 如果注册时被依赖的操作还未执行完，则由回调线程执行。

异步方法（即带Async后缀的方法）：可以选择是否传递线程池参数Executor运行在指定线程池中；当不传递Executor时，会使用ForkJoinPool中的共用线程池CommonPool（CommonPool的大小是CPU核数-1，如果是IO密集的应用，线程数可能成为瓶颈）。

例如：

ExecutorService threadPool1 = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(100));
CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
    System.out.println("supplyAsync 执行线程：" + Thread.currentThread().getName());
    //业务操作
    return "";
}, threadPool1);
//此时，如果future1中的业务操作已经执行完毕并返回，则该thenApply直接由当前main线程执行；否则，将会由执行以上业务操作的threadPool1中的线程执行。
future1.thenApply(value -> {
    System.out.println("thenApply 执行线程：" + Thread.currentThread().getName());
    return value + "1";
});
//使用ForkJoinPool中的共用线程池CommonPool
future1.thenApplyAsync(value -> {
//do something
  return value + "1";
});
//使用指定线程池
future1.thenApplyAsync(value -> {
//do something
  return value + "1";
}, threadPool1);

前面提到，异步回调方法可以选择是否传递线程池参数Executor，这里我们建议强制传线程池，且根据实际情况做线程池隔离。

当不传递线程池时，会使用ForkJoinPool中的公共线程池CommonPool，这里所有调用将共用该线程池，核心线程数=处理器数量-1（单核核心线程数为1），所有异步回调都会共用该CommonPool，核心与非核心业务都竞争同一个池中的线程，很容易成为系统瓶颈。手动传递线程池参数可以更方便的调节参数，并且可以给不同的业务分配不同的线程池，以求资源隔离，减少不同业务之间的相互干扰。

public Object doGet() {
  ExecutorService threadPool1 = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(100));
  CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
  //do sth
    return CompletableFuture.supplyAsync(() -> {
        System.out.println("child");
        return "child";
      }, threadPool1).join();//子任务
    }, threadPool1);
  return cf1.join();
}

如上代码块所示，doGet方法第三行通过supplyAsync向threadPool1请求线程，并且内部子任务又向threadPool1请求线程。threadPool1大小为10，当同一时刻有10个请求到达，则threadPool1被打满，子任务请求线程时进入阻塞队列排队，但是父任务的完成又依赖于子任务，这时由于子任务得不到线程，父任务无法完成。主线程执行cf1.join()进入阻塞状态，并且永远无法恢复。

为了修复该问题，需要将父任务与子任务做线程池隔离，两个任务请求不同的线程池，避免循环依赖导致的阻塞。

服务异步化后很多步骤都会依赖于异步RPC调用的结果，这时需要特别注意一点，如果是使用基于NIO（比如Netty）的异步RPC，则返回结果是由IO线程负责设置的，即回调方法由IO线程触发，CompletableFuture同步回调（如thenApply、thenAccept等无Async后缀的方法）如果依赖的异步RPC调用的返回结果，那么这些同步回调将运行在IO线程上，而整个服务只有一个IO线程池，这时需要保证同步回调中不能有阻塞等耗时过长的逻辑，否则在这些逻辑执行完成前，IO线程将一直被占用，影响整个服务的响应。

由于异步执行的任务在其他线程上执行，而异常信息存储在线程栈中，因此当前线程除非阻塞等待返回结果，否则无法通过trycatch捕获异常。CompletableFuture提供了异常捕获回调exceptionally，相当于同步调用中的trycatch。使用方法如下所示：

@Autowired
private WmOrderAdditionInfoThriftService wmOrderAdditionInfoThriftService;//内部接口
public CompletableFuture getCancelTypeAsync(long orderId) {
    CompletableFuture remarkResultFuture = wmOrderAdditionInfoThriftService.findOrderCancelledRemarkByOrderIdAsync(orderId);//业务方法，内部会发起异步rpc调用
    return remarkResultFuture
      .exceptionally(err -> {//通过exceptionally 捕获异常，打印日志并返回默认值
         log.error("WmOrderRemarkService.getCancelTypeAsync Exception orderId={}", orderId, err);
         return 0;
      });
}

有一点需要注意，CompletableFuture在回调方法中对异常进行了包装。大部分异常会封装成CompletionException后抛出，真正的异常存储在cause属性中，因此如果调用链中经过了回调方法处理那么就需要用Throwable.getCause()方法提取真正的异常。但是，有些情况下会直接返回真正的异常（Stack Overflow的讨论），最好使用工具类提取异常，如下代码所示：

@Autowired
private WmOrderAdditionInfoThriftService wmOrderAdditionInfoThriftService;//内部接口
public CompletableFuture getCancelTypeAsync(long orderId) {
    CompletableFuture remarkResultFuture = wmOrderAdditionInfoThriftService.findOrderCancelledRemarkByOrderIdAsync(orderId);//业务方法，内部会发起异步rpc调用
    return remarkResultFuture
          .thenApply(result -> {//这里增加了一个回调方法thenApply，如果发生异常thenApply内部会通过new CompletionException(throwable) 对异常进行包装
      //这里是一些业务操作
        })
      .exceptionally(err -> {//通过exceptionally 捕获异常，这里的err已经被thenApply包装过，因此需要通过Throwable.getCause()提取异常
         log.error("WmOrderRemarkService.getCancelTypeAsync Exception orderId={}", orderId, ExceptionUtils.extractRealException(err));
         return 0;
      });
}

上面代码中用到了一个自定义的工具类ExceptionUtils，用于CompletableFuture的异常提取，在使用CompletableFuture做异步编程时，可以直接使用该工具类处理异常。实现代码如下：

public class ExceptionUtils {
    public static Throwable extractRealException(Throwable throwable) {
          //这里判断异常类型是否为CompletionException、ExecutionException，如果是则进行提取，否则直接返回。
        if (throwable instanceof CompletionException || throwable instanceof ExecutionException) {
            if (throwable.getCause() != null) {
                return throwable.getCause();
            }
        }
        return throwable;
    }
}

@FunctionalInterface
public interface ThriftAsyncCall {
    void invoke() throws TException ;
}

@Slf4j
public class FutureUtils {

public static  CompletableFuture toCompletableFuture(final OctoThriftCallback callback , ThriftAsyncCall thriftCall) {
    CompletableFuture thriftResultFuture = new CompletableFuture<>();
    callback.addObserver(new OctoObserver() {
        @Override
        public void onSuccess(T t) {
            thriftResultFuture.complete(t);
        }
        @Override
        public void onFailure(Throwable throwable) {
            thriftResultFuture.completeExceptionally(throwable);
        }
    });
    if (thriftCall != null) {
        try {
            thriftCall.invoke();
        } catch (TException e) {
            thriftResultFuture.completeExceptionally(e);
        }
    }
    return thriftResultFuture;
}
  
  public static  CompletableFuture failed(Throwable ex) {
   CompletableFuture completableFuture = new CompletableFuture<>();
   completableFuture.completeExceptionally(ex);
   return completableFuture;
  }
  
  public static  CompletableFuture success(T result) {
   CompletableFuture completableFuture = new CompletableFuture<>();
   completableFuture.complete(result);
   return completableFuture;
  }
  
  public static  CompletableFuture> sequence(Collection> completableFutures) {
   return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0]))
           .thenApply(v -> completableFutures.stream()
                   .map(CompletableFuture::join)
                   .collect(Collectors.toList())
           );
  }
  
  public static  CompletableFuture> sequenceList(Collection>> completableFutures) {
   return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0]))
           .thenApply(v -> completableFutures.stream()
                   .flatMap( listFuture -> listFuture.join().stream())
                   .collect(Collectors.toList())
           );
  }
  
  public static  CompletableFuture> sequenceMap(
       Collection>> completableFutures, BinaryOperator mergeFunction) {
   return CompletableFuture
           .allOf(completableFutures.toArray(new CompletableFuture[0]))
           .thenApply(v -> completableFutures.stream().map(CompletableFuture::join)
                   .flatMap(map -> map.entrySet().stream())
                   .collect(Collectors.toMap(Entry::getKey, Entry::getValue, mergeFunction)));
  }
  
  public static  CompletableFuture> sequenceNonNull(Collection> completableFutures) {
   return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0]))
           .thenApply(v -> completableFutures.stream()
                   .map(CompletableFuture::join)
                   .filter(e -> e != null)
                   .collect(Collectors.toList())
           );
  }
  
  public static  CompletableFuture> sequenceListNonNull(Collection>> completableFutures) {
   return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0]))
           .thenApply(v -> completableFutures.stream()
                   .flatMap( listFuture -> listFuture.join().stream().filter(e -> e != null))
                   .collect(Collectors.toList())
           );
  }
  
  public static  CompletableFuture> sequence(Collection> completableFutures,
                                                     Predicate filterFunction) {
   return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0]))
           .thenApply(v -> completableFutures.stream()
                   .map(CompletableFuture::join)
                   .filter(filterFunction)
                   .collect(Collectors.toList())
           );
  }
  
  public static  CompletableFuture> sequenceList(Collection>> completableFutures,
                                                         Predicate filterFunction) {
   return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture[0]))
           .thenApply(v -> completableFutures.stream()
                   .flatMap( listFuture -> listFuture.join().stream().filter(filterFunction))
                   .collect(Collectors.toList())
           );
  }

  public static  CompletableFuture> sequenceMap(
       Collection>> completableFutures) {
   return CompletableFuture
           .allOf(completableFutures.toArray(new CompletableFuture[0]))
           .thenApply(v -> completableFutures.stream().map(CompletableFuture::join)
                   .flatMap(map -> map.entrySet().stream())
                   .collect(Collectors.toMap(Entry::getKey, Entry::getValue, (a, b) -> b)));
  }}

  public class ExceptionUtils {
   
   public static Throwable extractRealException(Throwable throwable) {
       if (throwable instanceof CompletionException || throwable instanceof ExecutionException) {
           if (throwable.getCause() != null) {
               return throwable.getCause();
           }
       }
       return throwable;
   }
  }

@Slf4j
  public abstract class AbstractLogAction {
  protected final String methodName;
  protected final Object[] args;
public AbstractLogAction(String methodName, Object... args) {
    this.methodName = methodName;
    this.args = args;
}
protected void logResult(R result, Throwable throwable) {
    if (throwable != null) {
        boolean isBusinessError = throwable instanceof TBase || (throwable.getCause() != null && throwable
                .getCause() instanceof TBase);
        if (isBusinessError) {
            logBusinessError(throwable);
        } else if (throwable instanceof DegradeException || throwable instanceof DegradeRuntimeException) {//这里为内部rpc框架抛出的异常，使用时可以酌情修改
            if (RhinoSwitch.getBoolean("isPrintDegradeLog", false)) {
                log.error("{} degrade exception, param:{} , error:{}", methodName, args, throwable);
            }
        } else {
            log.error("{} unknown error, param:{} , error:{}", methodName, args, ExceptionUtils.extractRealException(throwable));
        }
    } else {
        if (isLogResult()) {
            log.info("{} param:{} , result:{}", methodName, args, result);
        } else {
            log.info("{} param:{}", methodName, args);
        }
    }
}
private void logBusinessError(Throwable throwable) {
    log.error("{} business error, param:{} , error:{}", methodName, args, throwable.toString(), ExceptionUtils.extractRealException(throwable));
}
private boolean isLogResult() {
      //这里是动态配置开关，用于动态控制日志打印，开源动态配置中心可以使用nacos、apollo等，如果项目没有使用配置中心则可以删除
    return RhinoSwitch.getBoolean(methodName + "_isLogResult", false);
}}

@Slf4j
public class LogErrorAction extends AbstractLogAction implements BiConsumer {
public LogErrorAction(String methodName, Object... args) {
    super(methodName, args);
}
@Override
public void accept(R result, Throwable throwable) {
    logResult(result, throwable);
}
}

completableFuture
.whenComplete(
  new LogErrorAction<>("orderService.getOrder", params));

public class DefaultValueHandle extends AbstractLogAction implements BiFunction {
    private final R defaultValue;

private final boolean isNullToDefault;

  public DefaultValueHandle(String methodName, R defaultValue, Object... args) {
   super(methodName, args);
   this.defaultValue = defaultValue;
   this.isNullToDefault = false;
  }

  public DefaultValueHandle(boolean isNullToDefault, R defaultValue, String methodName, Object... args) {
   super(methodName, args);
   this.defaultValue = defaultValue;
   this.isNullToDefault = isNullToDefault;
  }
@Override
public R apply(R result, Throwable throwable) {
    logResult(result, throwable);
    if (throwable != null) {
        return defaultValue;
    }
    if (result == null && isNullToDefault) {
        return defaultValue;
    }
    return result;
}
public static  DefaultValueHandle.DefaultValueHandleBuilder builder() {
    return new DefaultValueHandle.DefaultValueHandleBuilder<>();
}
public static class DefaultValueHandleBuilder {
    private boolean isNullToDefault;
    private R defaultValue;
    private String methodName;
    private Object[] args;
    DefaultValueHandleBuilder() {
    }
    public DefaultValueHandle.DefaultValueHandleBuilder isNullToDefault(final boolean isNullToDefault) {
        this.isNullToDefault = isNullToDefault;
        return this;
    }
    public DefaultValueHandle.DefaultValueHandleBuilder defaultValue(final R defaultValue) {
        this.defaultValue = defaultValue;
        return this;
    }
    public DefaultValueHandle.DefaultValueHandleBuilder methodName(final String methodName) {
        this.methodName = methodName;
        return this;
    }
    public DefaultValueHandle.DefaultValueHandleBuilder args(final Object... args) {
        this.args = args;
        return this;
    }
    public DefaultValueHandle build() {
        return new DefaultValueHandle(this.isNullToDefault, this.defaultValue, this.methodName, this.args);
    }
    public String toString() {
        return "DefaultValueHandle.DefaultValueHandleBuilder(isNullToDefault=" + this.isNullToDefault + ", defaultValue=" + this.defaultValue + ", methodName=" + this.methodName + ", args=" + Arrays.deepToString(this.args) + ")";
    }
}

completableFuture.handle(new DefaultValueHandle<>("orderService.getOrder", Collections.emptyMap(), params));