ForkJoin框架之CompletableFuture

摘要：内部类，用于对和异常进行包装，从而保证对进行只有一次成功。是取消异常，转换后抛出。判断是否使用的线程池，在中持有该线程池的引用。

近期作者对响应式编程越发感兴趣，在内部分享"JAVA9-12"新特性过程中，有两处特性让作者深感兴趣：
1.JAVA9中的JEP266对并发编程工具的更新，包含发布订阅框架Flow和CompletableFuture加强，其中发布订阅框架以java.base模块下的java.util.concurrent.Flow及其中的几个内部类/接口为组成部分，它们的名称和作用如下，摘自JAVA12的Flow api文档。

2.JAVA9中孵化，JAVA11中标准化的HttpClient，在之前分享的JAVA9-12新特性一文中曾引用摘自网络的HttpClient代码片段：
片段1：

HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
    .uri(URI.create(uri))
    .build();

return client.sendAsync(request, HttpResponse.BodyHandlers.ofString())
    .thenApply(HttpResponse::body);
}

片段2：

HttpClient client = HttpClient.newHttpClient();
List urls = List.of("http://www.baidu.com","http://www.alibaba.com/","http://www.tencent.com");
List requests = urls.stream()
    .map(url -> HttpRequest.newBuilder(URI.create(url)))
    .map(reqBuilder -> reqBuilder.build())
    .collect(Collectors.toList());

List>> futures = requests.stream()
    .map(request -> client.sendAsync(request, HttpResponse.BodyHandlers.ofString()))
    .collect(Collectors.toList());
futures.stream()
    .forEach(e -> e.whenComplete((resp,err) -> {
        if(err != null){
            err.printStackTrace();
        }else{
            System.out.println(resp.body());
            System.out.println(resp.statusCode());
        }
    }));
CompletableFuture.allOf(futures
    .toArray(CompletableFuture[]::new))
    .join();
}

在片段1中，thenApply方法是CompletableFuture的成员，client.sendAsync返回的是一个CompletableFuture。这两段代码很好阅读，甚至说猜出其中的意义。片段2可以说对于作者目前的书写习惯是一个全面的颠覆，显然我们可以预先定义响应行为，而行为的执行时间则由前一个阶段的实际完成时间决定。片段2中的whenComplete方法很好理解，最后一行用allOf生成一个类似树的依赖结构，在当前方法中等待所有CompletableFuture执行完成。

简单看这两段代码，响应式编程的魅力可见一斑，甚至可以说是美妙不可言。
那么，作为JAVA9中额外照顾增强，HttpClient赖以实现的CompletableFuture，它是何方神圣呢？

CompletionStage是什么？不妨卖个关子先。
作者目前使用的JDK版本为8，尽管它不包含9之后的增强，万幸CompletionStage是从JDK8引入，因此足以用以了解这一伟大神器了。近期作者在公司使用的一些开源框架中，发现至处间接对它的使用：
1.持久化框架Redission。它内部使用一个RedissonExecutorService(实现ScheduledExecutorService）和PromiseDelegator（实现CompletionStage，而CompletableFuture同样也实现了CompletionStage）来异步地执行task。
2.apollo配置中心。它提供了配置变更的异步通知机制，而这依赖于spring web-mvc提供的DeferredResult，而在异步处理return value时，DeferredResult的setResult同样也是相应的CompletionStage执行。

//例：阿波罗NotificationControllerV2拉取通知接口
@GetMapping
public DeferredResult>>     pollNotification(
  @RequestParam(value = "appId") String appId,
  @RequestParam(value = "cluster") String cluster,
  @RequestParam(value = "notifications") String notificationsAsString,
  @RequestParam(value = "dataCenter", required = false) String dataCenter,
  @RequestParam(value = "ip", required = false) String clientIp) {
List notifications = null;
//省略无关代码
//DeferredResultWrapper是apollo作者包装的spring DeferredResult
DeferredResultWrapper deferredResultWrapper = new DeferredResultWrapper();
//省略无关代码
if (!CollectionUtils.isEmpty(newNotifications)) {
  deferredResultWrapper.setResult(newNotifications);
} else {
  deferredResultWrapper
      .onTimeout(() -> logWatchedKeys(watchedKeys, "Apollo.LongPoll.TimeOutKeys"));
 
  deferredResultWrapper.onCompletion(() -> {
    //unregister all keys
    for (String key : watchedKeys) {
      deferredResults.remove(key, deferredResultWrapper);
    }
    logWatchedKeys(watchedKeys, "Apollo.LongPoll.CompletedKeys");
  });
 
  //省略
return deferredResultWrapper.getResult();
}

在spring的CompletionStageReturnValueHandler的handleReturnValue（）方法中，如下异步地处理响应结果：

@Override
public void handleReturnValue(Object returnValue, MethodParameter returnType,
        ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {
 
    if (returnValue == null) {
        mavContainer.setRequestHandled(true);
        return;
    }
 
    final DeferredResult