《Java8实战》读书笔记10：组合式异步编程 CompletableFuture

《Java8实战》读书笔记10：组合式异步编程 CompletableFuture

• 第11章 CompletableFuture：组合式异步编程
• 11.1 Future 接口 （只是个引子）
• • 11.1.1 Future 接口的局限性
  • 11.1.2 使用 CompletableFuture 构建异步应用
• 11.2 实现异步 API
• • 11.2.0 同步版本 Shop 类（用来重构的原料）
  • 11.2.1 将同步方法转换为异步方法
  • 11.2.2 错误处理
  • CompletableFuture 静态方法 supplyAsync（最终版）
• 11.3 让你的代码免受阻塞之苦
• • 11.3.0 顺序流操版
  • 11.3.1 使用并行流对请求进行并行操作（并行流版）
  • 11.3.2 使用 CompletableFuture 发起异步请求（过度版）
  • 11.3.3 寻找更好的方案
  • 11.3.4 使用定制的执行器（最终版）
  • • 自定义执行器 Executor
    • 使用自定义执行器
    • 测试性能
    • 我们对使用这些API的建议如下
• 11.4 对多个异步任务进行流水线操作 （实现串连并连等）
• • 11.4.0-1 准备工作（同步版）
  • • 实现价格解析类 Quote
    • 实现折扣服务 Discount
    • 实现商店类 Shop
    • 测试类 ShopTest
  • 11.4.2 使用 Discount 服务
  • 11.4.3 构造同步和异步操作
    （thenCompose 前任务结果作为后任务参数）
  • 11.4.4 将两个 CompletableFuture ...
    （thenCombine 两个都完成再一起处理）
  • 11.4.5 对 Future 和 CompletableFuture 的回顾（Java 7）
• 11.5 响应 CompletableFuture 的 completion 事件（回调函数）
• • 11.5.1 对最佳价格查询器应用的优化
  • 11.5.2 付诸实践 thenAccept
  • allOf 所有都完成就结束
  • anyOf 任意一个完成就结束
• 11.6 小结
• CompletableFuture 部分方法介绍
• • acceptEither 谁先完成用谁
  • thenAccept 完成后消费结果
  • handle 完成后加工结果并返回
  • obtrudeValue 直接设置一个值为结果
  • thenAcceptBoth 两个都完成就消费
  • whenComplete 完成后返回前，可读取结果
  • exceptionally 如果异常返回给定值，否则正常返回
• 参考资料

第11章 CompletableFuture：组合式异步编程

本章内容
 创建异步计算，并获取计算结果
 使用非阻塞操作提升吞吐量
 设计和实现异步API
 如何以异步的方式使用同步的API
 如何对两个或多个异步操作进行流水线和合并操作
 如何处理异步操作的完成状态

1. 看了一下 本章内容 我感觉到有点闪回的效果，怎么似曾相识呢。。。哦！！！
   1.1 前面的《Java8实战》读书笔记03：Lambda 表达式的组合用法 就有类似的概念。
   1.2. 之前的JS中也学了类似的东西：JavaScript 学习笔记【Promise】async、await
2. 除非是使用 消息队列 进行异步解耦。否则大多数情况异步任务存在的意义就是别耽误主线程干活（异步任务是被老大派出去办事的小弟），主线程总会有那么一个时间点，堵在校门口让异步任务把结果交出来，供主线程享用。

11.1 Future 接口 （只是个引子）

1. Future 接口在Java 5中被引入。它代表了一个对将来的结果的引用，并提供了可获取最终结果的方法；另一个优点是它比更底层的 Thread 易用。
2. 从下面代码中可以看到它需要配合 ExecutorService 线程池使用。
3. Future 的 get 方法：如果操作已经完成，该方法会立刻返回操作的结果，否则它会阻塞你的线程，直到操作完成，返回相应的结果。它还有一个可以设置超时重载版本。（除非你正在被迫为敌国写代码，否则，请始终设置超时时间）

public static void main(String[] args) {
    //System.out.println("1. 主线程开始...");
    ExecutorService executor = Executors.newCachedThreadPool();
    //System.out.println("2. 提交异步任务!");
    Future future = executor.submit(new Callable() {
        public Double call() {
            return doSomeLongComputation();
        }});
    doSomethingElse();
    try {
        //System.out.println("4. 主线程忙完了，向异步任务要结果，等到它返回为止。");
        Double result = future.get(2000, TimeUnit.SECONDS);
        //System.out.println("5. 得到异步结果: " + result);
    } catch (ExecutionException ee) {}
    catch (InterruptedException ie) {}
    catch (TimeoutException te) {}
    //System.out.println("6. 完事...");
    //executor.shutdown();
}
// 某耗时运算
private static Double doSomeLongComputation(){
    try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); }
    return 0.618D;
}
// 主线程干其它活
private static void doSomethingElse(){
    System.out.println("3. 主线程继续忙别的事件");
}

11.1.1 Future 接口的局限性

看看本章的标题，也能猜出Future只是个引子，它虽然也提供了一个 boolean isDone() 用来判断异步任务是否完成。但总的来说还是太简陋。

• 比如我们需要下面这些：

1. 将两个异步计算合并为一个——这两个异步计算之间相互独立，同时第二个又依赖于第一个的结果。(串连两个异步任务)
2. 等待Future集合中的所有任务都完成。
3. 仅等待Future集合中最快结束的任务完成（有可能因为它们试图通过不同的方式计算同一个值），并返回它的结果。
4. 通过编程方式完成一个Future任务的执行（即以手工设定异步操作结果的方式）。
5. 应对Future的完成事件（即当Future的完成事件发生时会收到通知，并能使用Future计算的结果进行下一步的操作，不只是简单地阻塞等待操作的结果）。

接下来，了解新的 CompletableFuture 类（它实现了 Future接口） 如何利用 Java 8 的新特性以更直观的方式将上述需求都变为可能。Stream 和 CompletableFuture 的设计都遵循了类似的模式：它们都使用了 Lambda 表达式以及 流水线 的思想。从这个角度，你可以说 CompletableFuture 和 Future 的关系就跟 Stream 和 Collection 的关系一样。

11.1.2 使用 CompletableFuture 构建异步应用

本节说后续会做一个“最佳价格查询器”，从中学到不拉不拉（直接看下文即可）略。。。

同步API与异步API
同步API其实只是对传统方法调用的另一种称呼：你调用了某个方法，调用方在被调用方运行的过程中会等待，被调用方运行结束返回，调用方取得被调用方的返回值并继续运行。即使调用方和被调用方在不同的线程中运行，调用方还是需要等待被调用方结束运行，这就是阻塞式调用这个名词的由来。
与此相反，异步API会直接返回，或者至少在被调用方计算完成之前，将它剩余的计算任务交给另一个线程去做，该线程和调用方是异步的——这就是非阻塞式调用的由来。执行剩余计算任务的线程会将它的计算结果返回给调用方。返回的方式：
——要么是通过回调函数，
——要么是由调用方再次执行一个“等待，直到计算完成”的方法调用。
这种方式的计算在I/O系统程序设计中非常常见：你发起了一次磁盘访问，这次访问和你的其他计算操作是异步的，你完成其他的任务时，磁盘块的数据可能还没载入到内存，你只需要等待数据的载入完成。

11.2 实现异步 API 11.2.0 同步版本 Shop 类（用来重构的原料）

先准备一份 同步版本 代码，然后再一步一步重构最终得到一个异步非堵塞的版本。

代码清单11-2 模拟1秒钟延迟的方法

public static void delay() {
    try { Thread.sleep(1000L); } 
    catch (InterruptedException e) { throw new RuntimeException(e); }
}

代码清单11-3 在getPrice方法中引入一个模拟的延迟

// 获取价格
public double getPrice(String product) {
    return calculatePrice(product);
}
// 计算价格
private double calculatePrice(String product) {
    delay();
    return new Random().nextDouble() * product.charAt(0) + product.charAt(1);
}

11.2.1 将同步方法转换为异步方法

将 getPrice 转换为 getPriceAsync 方法。它无法直接返回结果，但可以给你打个白条，等他有结果了，你可以拿着白条找他兑现。Lannister always pay his debts.

代码清单11-4 getPriceAsync方法的实现

// 异步获取价格
public Future getPriceAsync(String product) {
    CompletableFuture futurePrice = new CompletableFuture<>(); // 创建“白条”
    new Thread( () -> {
        double price = calculatePrice(product); // 在另一个线程中以异步方式执行计算
        futurePrice.complete(price); // 耗时任务结束，结果装进“白条”
    }).start();
    return futurePrice; // 返回“白条”
}

代码清单11-5 使用异步API (测试)

@Test
public void testGetPriceAsync() {
    Shop shop = new Shop("螺蛳粉店");
    long start = System.nanoTime();
    // 这里的 futurePrice 不是价格本身，而是一个在将来能取出 “螺蛳粉价格” 的纳戒。
    Future futurePrice = shop.getPriceAsync("螺蛳粉");
    long invocationTime = ((System.nanoTime() - start) / 1_000_000);
    System.out.println("发起异步调用在 " + invocationTime + " 毫秒后返回主线程。");
    // 主线各执行其它任务，比如
    System.out.println("猛刷外卖APP看还有别的什么东西吃。");
    // 在计算商品价格的同时
    try {
        // 从 Future（纳戒）中取价格，如果价格还没算好，就死等。
        double price = futurePrice.get();
        System.out.printf("价格是： %.2f%n", price);
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
    long retrievalTime = ((System.nanoTime() - start) / 1_000_000);
    System.out.println( retrievalTime + " 毫秒后返回价格。");
}

发起异步调用在 39 毫秒后返回主线程。
猛刷外卖APP看还有别的什么东西吃。
价格是： 43386.46
1047 毫秒后返回价格。

11.2.2 错误处理

上面的代码中计算价格如果发生异常，它可是在另一个线程啊，要怎么处理呢？接着看：

1. 改造 getPriceAsync 把可能出问题的计算价格过程try起来，
2. 如果有异常，用 futurePrice.completeExceptionally(e) 接住。
3. 如此客户端 futurePrice.get(); 时就会抛这个异常了。然后就看业务，处理就是了。（本质上就是把异常通过CompletableFuture 转了一手，带到主线程来了。）

代码清单11-6 抛出CompletableFuture内的异常

// 异步获取价格
public Future getPriceAsync(String product) {
    CompletableFuture futurePrice = new CompletableFuture<>(); // 创建 CompletableFuture 对象，它会包含计算的结果
    new Thread( () -> {
        try {
            double price = calculatePrice(product); // 在另一个线程中以异步方式执行计算
            futurePrice.complete(price); // 耗时任务结束，结果装进“白条”
        } catch (Exception e) {
        	// 设置异常后，再调用  get() 时就会抛它出来。
            futurePrice.completeExceptionally(e); 
        }
    }).start();
    return futurePrice; // 返回“白条”
}

CompletableFuture 静态方法 supplyAsync（最终版）

• 使用工厂方法supplyAsync创建CompletableFuture

CompletableFuture类自身提供了大量精巧的工厂方法，使用这些方法能更容易地完成整个流程，还不用担心实现的细节。比如，采用supplyAsync方法后，你可以用一行语句重写代码清单11-4中的getPriceAsync方法，如下所示。

代码清单11-7 使用工厂方法supplyAsync创建CompletableFuture对象

public Future getPriceAsync(String product) { 
	return CompletableFuture.supplyAsync(() -> calculatePrice(product)); 
}

这一句就等价于 代码清单11-6 那一堆。（包括异常处理的部分） 所以后续应用场景，我们肯定是直接上这个版本了。

11.3 让你的代码免受阻塞之苦 11.3.0 顺序流操版

前面实现了单个商店的查询，但是如果有多个商店需要查询，问题就有点不同了。
现在的剧情是：你需要查询 所有商店，但他们都只提供了同步API（getPrice方法）

代码清单11-8 采用顺序查询所有商店的方式实现的findPrices方法 （批量查询）

// 批量查询价格
public static List findPrices(String product) {
    return shops.stream()
            .map(shop -> String.format("%s 价格为： %.2f", 
				            shop.getName(), 
				            shop.getPrice(product)))
            .collect(toList());
}

代码清单11-9 验证正确性和执行性能

// 测试数据【商家的列表】，如下：
private static List shops = Arrays.asList(
        new Shop("柳州螺蛳粉"), new Shop("津市牛肉粉"),
        new Shop("重庆酸辣粉"), new Shop("桂林米粉"));

@Test
public void testFindPrices() {
    long start = System.nanoTime();
    List str = findPrices("酸辣粉");
    long duration = (System.nanoTime() - start) / 1_000_000;
    System.out.println("耗时 " + duration + " 纳秒");
}

[【柳州螺蛳粉】家【酸辣粉】价格： 72101.30, 【津市牛肉粉】家【酸辣粉】价格： 60069.58, 
【重庆酸辣粉】家【酸辣粉】价格： 72050.75, 【桂林米粉】家【酸辣粉】价格： 61647.51]
耗时 4084 纳秒

11.3.1 使用并行流对请求进行并行操作（并行流版）

代码清单11-8 中 shops.stream() 换 shops.parallelStream() 即可，直接看测试结果：

[【柳州螺蛳粉】家【酸辣粉】价格： 37838.66, 【津市牛肉粉】家【酸辣粉】价格： 65524.66, 
【重庆酸辣粉】家【酸辣粉】价格： 52324.70, 【桂林米粉】家【酸辣粉】价格： 60152.23]
耗时 1064 纳秒

效率有明显提升。

11.3.2 使用 CompletableFuture 发起异步请求（过度版）

代码清单11-11 使用 CompletableFuture 实现 findPrices 方法
注意 CompletableFuture 类中的 join 方法和 Future 接口中的 get 有相同的含义，并且也声明在 Future 接口中，它们唯一的不同是 join不会抛出任何检测到的异常。

public static List findPrices(String product) {
    List> priceFutures =
            shops.stream()
                    .map(shop -> CompletableFuture.supplyAsync(
                            () -> String.format("【%s】家【%s】价格： %.2f", shop.getName(), product, shop.getPrice(product))
                        )
                    )
                    .collect(Collectors.toList());
    return priceFutures.stream()
            .map(CompletableFuture::join)
            .collect(toList());
}

这里可以看到，代码中分成了两段流操作。对应 图11-4 的下半部分（并行执行）。
第一部分是顺序流，如果把第二部分直接接着它map 那就都成顺序流了。所以先把结果future1、future2、future3聚集到一个列表中，再开一个流。这样所有future的join()能并行启动，不用排队等前一个结束。

[【柳州螺蛳粉】家【酸辣粉】价格： 37810.56, 【津市牛肉粉】家【酸辣粉】价格： 58265.10, 
【重庆酸辣粉】家【酸辣粉】价格： 64577.53, 【桂林米粉】家【酸辣粉】价格： 51406.55]
耗时 1071 纳秒

这里效果还不如直接上面直接用并行流的方案。然后引出后续剧情。

11.3.3 寻找更好的方案

并行流与CompletableFuture内部默认用的都是通用线程池，固定数目的线程，具体线程数取决于 Runtime.getRuntime().availableProcessors() （Java虚拟机可用的处理器数量 ） 的返回值。

然而，CompletableFuture 允许你对执行器（Executor）进行配置，尤其是线程池的大小，让它以更适合应用需求的方式进行配置，满足程序的要求，而这是并行流API无法提供的。

11.3.4 使用定制的执行器（最终版）

调整线程池的大小
《Java并发编程实战》一书中，Brian Goetz和合著者们为线程池大小的优化提供了不少中肯的建议。这非常重要，如果线程池中线程的数量过多，最终它们会竞争稀缺的处理器和内存资源，浪费大量的时间在上下文切换上。反之，如果线程的数目过少，正如你的应用所面临的情况，处理器的一些核可能就无法充分利用。Brian Goetz建议，线程池大小与处理器的利用率之比可以使用下面的公式进行估算：
Nthreads = NCPU * UCPU * (1 + W/C)
其中：
❑NCPU是处理器的核的数目，可以通过 Runtime.getRuntime().availableProcessors() 得到
❑UCPU是期望的CPU利用率（该值应该介于0和1之间）
❑W/C是等待时间与计算时间的比率

简化方案：线程数 = 商店数，再加个上限保底100：Math.min(shops.size(), 100)

自定义执行器 Executor

代码清单11-12 为“最优价格查询器”应用定制的执行器

private final Executor executor = Executors.newFixedThreadPool(
		Math.min(shops.size(), 100),
       	new ThreadFactory() {
        	public Thread newThread(Runnable r) {
				Thread t = new Thread(r);
				t.setDaemon(true); // 使用守护线程——这种方式不会阻止程序的关停
				return t;
			}
       }
   );

还记得用 lambda 替换 匿名类 吗？ lambda 版：

private static final Executor executor = Executors.newFixedThreadPool(
        Math.min(shops.size(), 100),
        (r) ->{
            Thread t = new Thread(r);
            t.setDaemon(true); // 使用守护线程——这种方式不会阻止程序的关停
            return t;
        }
);

果将线程标记为守护进程，意味着程序退出时它也会被回收。

使用自定义执行器

代码清单11-11 中 CompletableFuture.supplyAsync 加上第二个参数 executor，即可：

CompletableFuture.supplyAsync(
	() -> String.format("【%s】家【%s】价格： %.2f", shop.getName(), product, shop.getPrice(product)),
	executor
)

测试性能

[【柳州螺蛳粉0】家【酸辣粉】价格： 44212.36, 【柳州螺蛳粉1】家【酸辣粉】价格： 60541.35, 
【柳州螺蛳粉2】家【酸辣粉】价格： 62474.67, 【柳州螺蛳粉3】家【酸辣粉】价格： 56118.18, 
【柳州螺蛳粉4】家【酸辣粉】价格： 53683.30, 【柳州螺蛳粉5】家【酸辣粉】价格： 44924.01, 
【柳州螺蛳粉6】家【酸辣粉】价格： 67151.76, 【柳州螺蛳粉7】家【酸辣粉】价格： 59344.12, 
【柳州螺蛳粉8】家【酸辣粉】价格： 44457.99, 【柳州螺蛳粉9】家【酸辣粉】价格： 67508.70, 
【津市牛肉粉1】家【酸辣粉】价格： 54769.93, 【津市牛肉粉2】家【酸辣粉】价格： 63197.59, 
【重庆酸辣粉1】家【酸辣粉】价格： 73413.08, 【重庆酸辣粉2】家【酸辣粉】价格： 38340.71, 
【桂林米粉】家【酸辣粉】价格： 57547.14]
耗时 1064 纳秒

这里把商店添加到了 15 家。耗时反而最少。性能满意
一般而言，这种状态会一直持续，直到商店的数目达到我们之前计算的阈值(程池大小)。

并行——使用流还是CompletableFutures？
目前为止，你已经知道对集合进行并行计算有两种方式：
: 要么将其转化为并行流，利用map这样的操作开展工作，
: 要么枚举出集合中的每一个元素，创建新的线程，在 CompletableFuture 内对其进行操作。后者提供了更多的灵活性，你可以调整线程池的大小，而这能帮助你确保整体的计算不会因为线程都在等待I/O而发生阻塞。

我们对使用这些API的建议如下

❑ 如果你进行的是计算密集型的操作，并且没有I/O，那么推荐使用Stream接口，因为实现简单，同时效率也可能是最高的（如果所有的线程都是计算密集型的，那就没有必要创建比处理器核数更多的线程）。
❑ 反之，如果你并行的工作单元还涉及等待I/O的操作（包括网络连接等待），那么使用 CompletableFuture 灵活性更好，你可以像前文讨论的那样，依据等待/计算，或者W/C的比率设定需要使用的线程数。这种情况不使用并行流的另一个原因是，处理流的流水线中如果发生I/O等待，流的延迟特性会让我们很难判断到底什么时候触发了等待。

11.4 对多个异步任务进行流水线操作 （实现串连并连等）

剧情：现在除了查价格还要查折扣，然后再算出实付多少。
首先准备一个同步版，然后再慢慢重构。

11.4.0-1 准备工作（同步版）

现在我们需要添加几个类一起来配合，最终：（代码稍微整理了一下， 完全照抄）
Quote：价格解析类【新增】
Discount：价格解析类【新增】
Shop：这就是前面的商店类。
ShopTest：测试类。

实现价格解析类 Quote

封装一个 Quote 它用静态方法 parse 解析 getPrice 返回的字符串结果：商品名称:价格:折扣码。得到三个值后，new一个 Quote 实例返回。
原文较长，我这里顺便用 lombok 简化了一下。

@Getter
@AllArgsConstructor
public class Quote {
    private final String shopName;
    private final double price;
    private final Discount.Code discountCode;

    public static Quote parse(String s) {
        String[] split = s.split(":");
        String shopName = split[0];
        double price = Double.parseDouble(split[1]);
        Discount.Code discountCode = Discount.Code.valueOf(split[2]);
        return new Quote(shopName, price, discountCode);
    }
}

实现折扣服务 Discount

内用枚举实现的.

代码清单11-13 以枚举类型定义的折扣代码

public class Discount {
    @AllArgsConstructor
    public enum Code {
        NONE(0, "原价"),
        SILVER(5, "白银9.5折"), GOLD(10, "黄金9折"),
        PLATINUM(15, "铂金8.5折"), DIAMOND(20, "钻石8折");
        private final int percentage;
        private final String name;
    }
    // 格式化折后价：店名-折扣 价格：XXXX
    public static String applyDiscount(Quote quote) {
        return quote.getShopName() + "-" + quote.getDiscountCode().name
                + " 价格：" + Discount.apply(quote.getPrice(), quote.getDiscountCode());
    }
    // 计算折后价
    private static double apply(double price, Code code) {
        try { Thread.sleep(1000L); } catch (InterruptedException e) { throw new RuntimeException(e); }
        return (price * (100 - code.percentage) / 100);
    }
}

实现商店类 Shop

修改getPrice方法的返回格式：商品名称:价格:折扣码。
折扣码 Discount.Code 随机生成
价格：由 calculatePrice 返回。

@Data
public class Shop {
    private String name;
    private Random random;

    public Shop(String bestShop) { name = bestShop; random = new Random(); }

    // 查询价格
    public String getPrice(String product) {
        double price = calculatePrice(product);
        // 共有5个折扣级别，随机获取。
        Discount.Code code = Discount.Code.values()[random.nextInt(Discount.Code.values().length)];
        // 调整返回格式为： 商品名称:价格:折扣码
        return String.format("%s:%.2f:%s", name, price, code);
    }
    // 异步获取价格
    public Future getPriceAsync(String product) {
        return CompletableFuture.supplyAsync(() -> calculatePrice(product));
    }
    // 模拟耗时操作,延迟1s
    public void delay() {
        try { Thread.sleep(1000L); } catch (InterruptedException e) { throw new RuntimeException(e); }
    }
    // 计算价格
    private double calculatePrice(String product) {
        delay(); // 延迟
        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }
}

测试类 ShopTest

public class ShopTest {
    // 测试数据
    private List shops = Arrays.asList(new Shop("柳州螺蛳粉"), new Shop("津市牛肉粉"),
        new Shop("重庆酸辣粉"), new Shop("过桥米线"), new Shop("桂林米粉"));

    // 批量查询价格
    public  List findPrices(String product) {
        return shops.stream()
                .map(shop ->shop.getPrice(product)) // 按产品名获取：价格、折扣码
                .map(Quote::parse) // 解析“商品名称:价格:折扣码”字符串
                .map(Discount::applyDiscount) // 算出折后价，并格式化为字符串
                .collect(toList());
    }
    // 批量查询价格
    @Test
    public void testFindPrices() {
        long start = System.nanoTime();
        List str = findPrices("酸辣粉");
        long duration = (System.nanoTime() - start) / 1_000_000;
        System.out.println("耗时 " + duration + " 纳秒");
    }
}    

11.4.2 使用 Discount 服务

执行上面的代码，直接看结果

柳州螺蛳粉-黄金9折 价格：49960.08900000001
津市牛肉粉-黄金9折 价格：52730.793
重庆酸辣粉-白银9.5折 价格：60903.246
过桥米线-铂金8.5折 价格：39914.5635
桂林米粉-白银9.5折 价格：60059.9025
耗时 10093 纳秒

11.4.3 构造同步和异步操作
（thenCompose 前任务结果作为后任务参数）

代码清单11-16 使用CompletableFuture实现findPrices方法
重构 findPrices 批量查询价格方法，执行器 使用的就是 代码清单11-12 的 executor
为了让流操作部分更加直观，我把两个 lambda 提了出来。
图11-5 构造同步操作和异步任务

public List findPrices(String product) {
    // 异步获取价格，返回 CompletableFuture
    Function> getPrice = (Shop shop) -> {
        return CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor);
    };
    // 异步获取折后价，返回 CompletableFuture
    Function> getDiscountPrice = (Quote quote) -> {
        return CompletableFuture.supplyAsync( () -> Discount.applyDiscount(quote), executor );
    };

    List> priceFutures =
        shops.stream()
            .map(getPrice) // 1.异步获取价格
            .map(future -> future.thenApply(Quote::parse)) // 2.解析价格字符串为 Quote
            .map(future -> future.thenCompose(getDiscountPrice)) // 3.异步获取折后价
            .collect(toList());
                
    return priceFutures.stream()
            .map(CompletableFuture::join)
            .collect(toList());
}

1. 使用工厂方法 CompletableFuture.supplyAsync 异步执行价格查询。返回 Stream> 对象，运行结束后每个 CompletableFuture 中都会包一份 shop.getPrice 返回的字符串。并且我们指定了执行器 executor。
2. 解析上一步得到的价格字符串，没有异步操作也没有IO，可以同步执行。对上一步生成的 CompletableFuture 对象调用它的thenApply，将 Quote::parse 作为参数传给它。
3. 第三个map操作涉及联系远程的Discount服务，这需要串连两个异步。可以用thenCompose实现。

// 批量查询价格
@Test
public void testFindPrices() {
    long start = System.nanoTime();
    findPrices("酸辣粉").forEach(System.out::println);
    long duration = (System.nanoTime() - start) / 1_000_000;
    System.out.println("耗时 " + duration + " 纳秒");
}

明显改善

柳州螺蛳粉-白银9.5折 价格：40852.2895
津市牛肉粉-白银9.5折 价格：68308.686
重庆酸辣粉-黄金9折 价格：59532.894
过桥米线-白银9.5折 价格：52773.373999999996
桂林米粉-铂金8.5折 价格：56312.91650000001
耗时 2024 纳秒

11.4.4 将两个 CompletableFuture …
（thenCombine 两个都完成再一起处理）

将两个 CompletableFuture 对象整合起来，无论它们是否存在依赖

代码清单11-17 合并两个独立的CompletableFuture对象
findPrices 的另一个版本，它的业务是：

1. 先查到价格。
2. 再查到汇率（这里我写死 0.67d了）
3. 拿到上面两个异步的结果，按业务需要处理得到最终结果。

图11-6 合并两个相互独立的异步任务

public List findPrices(String product) {
    // 异步获取价格
    Function> getPrice = (Shop shop) -> {
        return CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor);
    };
    // 与第二个异步任务合并
    Function, CompletableFuture> futureThenCombine = future -> {
        return future.thenCombine(
                CompletableFuture.supplyAsync(() -> 0.67d), // 得到：rate
                // price 是上个异步的结果，rate 是 thenCombine 中这个异步的结果
                (price, rate) -> price * rate
        );
    };

    return shops.stream()
            .map(getPrice) // 得到：price
            .map(futureThenCombine)
            .collect(toList()) // 收集为 list
            .stream() // 再生成一个新流
            .map(CompletableFuture::join)
            .map(String::valueOf)
            .collect(toList());
}

测试代码不变，直接看结果：

38381.82494202258
44299.93034171271
35652.96944201568
38036.79768005054
34276.16511900577
耗时 1024 纳秒

11.4.5 对 Future 和 CompletableFuture 的回顾（Java 7）

尝试仅使用Java 7中提供的特性，重新实现代码清单11-17的功能。
代码清单11-18 利用Java 7的方法合并两个Future对象
只实现了 thenCombine 部分：

@Test
public void testFindPricesJava7() throws ExecutionException, InterruptedException {
    Shop shop = new Shop("重庆酸辣粉");
    ExecutorService executor = Executors.newCachedThreadPool();
    final Future futureRate = executor.submit(new Callable() {
        public Double call() {
            return 0.67d; 
        }});
    Future futurePriceInUSD = executor.submit(new Callable() {
        public Double call() throws ExecutionException, InterruptedException {
            double priceInEUR = shop.getPrice("酸辣粉");
            return priceInEUR * futureRate.get();
        }});

    long start = System.nanoTime();
    System.out.println(futurePriceInUSD.get());
    long duration = (System.nanoTime() - start) / 1_000_000;
    System.out.println("耗时 " + duration + " 纳秒");
}

11.5 响应 CompletableFuture 的 completion 事件（回调函数）

接下来我们实现两种效果：

1. 查到价格就显示（因为耗时不同，所以会有先有后）。直到所有都查完。
2. 查询所有，只显示最快的那家，就结束。

为了演示这个效果，我们需要让各商店获取价格的耗时有长有短。

11.5.1 对最佳价格查询器应用的优化

1. 将计算价格 Shop.calculatePrice 中的 delay 换成随机延迟版 randomDelay。

// 计算价格
private double calculatePrice(String product) {
    randomDelay(); // 随机延迟
    return random.nextDouble() * product.charAt(0) + product.charAt(1);
}

2. 添加机延迟方法 Shop.randomDelay
   代码清单11-19 一个模拟生成0.5秒至2.5秒随机延迟的方法

public void randomDelay() {
    int delay = 500 + random.nextInt(2000);
    try { Thread.sleep(delay); } catch (InterruptedException e) { throw new RuntimeException(e); }
}

3. 重构 ShopTest.findPrices 为 ShopTest.findPricesStream
   代码清单11-20 重构 findPrices 方法返回一个由 Future 构成的流

public Stream> findPricesStream(String product) {
    // 异步获取价格
    Function> getPrice = (Shop shop) -> {
        return CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor);
    };
    // 异步获取折后价
    Function> getDiscountPrice = (Quote quote) -> {
        return CompletableFuture.supplyAsync( () -> Discount.applyDiscount(quote), executor );
    };
    return shops.stream()
            .map(getPrice)
            .map(future -> future.thenApply(Quote::parse))
            .map(future -> future.thenCompose(getDiscountPrice));
}

11.5.2 付诸实践 thenAccept allOf 所有都完成就结束

// 批量查询价格
@Test
public void testFindPricesStream() {
    long start = System.nanoTime(); // 开始时间
    // 回到函数，显示查询到的折后价字符串
    Consumer showLog = discountPrice -> {
        long duration = (System.nanoTime() - start) / 1_000_000 ; // 耗时
        System.out.println(String.format("%s 查询完成，耗时 %d 纳秒 ", discountPrice , duration));
    };

    CompletableFuture[] futures = findPricesStream("螺蛳粉")
            .map(future -> future.thenAccept(showLog))
            .toArray(size -> new CompletableFuture[size]);

    CompletableFuture.allOf(futures).join();

    System.out.println("查完所有商店耗时： " + ((System.nanoTime() - start) / 1_000_000) + " 纳秒");
}

查到就输出，快先慢后。最后全部完成，再结案陈词。

重庆酸辣粉-原价 价格：39682.05 查询完成，耗时 1964 纳秒 
过桥米线-原价 价格：36999.13 查询完成，耗时 2081 纳秒 
桂林米粉-铂金8.5折 价格：52832.46399999999 查询完成，耗时 2709 纳秒 
柳州螺蛳粉-白银9.5折 价格：32835.6575 查询完成，耗时 2766 纳秒 
津市牛肉粉-铂金8.5折 价格：46930.8715 查询完成，耗时 2916 纳秒 
查完所有商店耗时： 2916 纳秒

anyOf 任意一个完成就结束

只要改一句：

CompletableFuture.anyOf(futures).join();

只取最快的一个输出，其余的不要了。

柳州螺蛳粉-原价 价格：58316.76 查询完成，耗时 1573 纳秒 
查完所有商店耗时： 1573 纳秒

11.6 小结

这一章中，你学到的内容如下。

1. 执行比较耗时的操作时，尤其是那些依赖一个或多个远程服务的操作，使用异步任务可以改善程序的性能，加快程序的响应速度。
2. 你应该尽可能地为客户提供异步API。使用CompletableFuture类提供的特性，你能够轻松地实现这一目标。
3. CompletableFuture类还提供了异常管理的机制，让你有机会抛出/管理异步任务执行中发生的异常。
4. 将同步API的调用封装到一个CompletableFuture中，你能够以异步的方式使用其结果。
5. 如果异步任务之间相互独立，或者它们之间某一些的结果是另一些的输入，你可以将这些异步任务构造或者合并成一个。
6. 你可以为CompletableFuture注册一个回调函数，在Future执行完毕或者它们计算的结果可用时，针对性地执行一些程序。
7. 你可以决定在什么时候结束程序的运行，是等待由CompletableFuture对象构成的列表中所有的对象都执行完毕，还是只要其中任何一个首先完成就中止程序的运行。

CompletableFuture 部分方法介绍

访求	异步版	说明
acceptEither	有	a.acceptEither(b, Consumer) 返回一个新的CompletionStage，当 a或 b 谁先完成，就将结果作为参数执行Consumer。 有关异常完成的规则，请参阅CompletionStage文档。
allOf		所有的对象都执行完毕，中止程序运行。
anyOf		任何一个首先完成就中止程序运行。
applyToEither	有	a.applyToEither(b, Function) 参考 acceptEither 只一过这里的函数式接口是Function 可以有返回值。
cancel		a.cancel(boolean 运行时可能中断) 如果 a 尚未完成则抛 CancellationException 来完成任务。
complete		a.complete(给定值) 如果 a 尚未完成，则将get()和相关方法返回的值设置为给定值。
thenApply	有	a.thenApply(Consumer) 返回一个新的CompletionStage，当 a 完成时，将结果作为 Consumer 的参数执行。 有关异常完成的规则，请参阅CompletionStage文档。
thenCompose	有	a.thenCompose(Function) 返回一个新的CompletionStage，当 a 完成时，将结果作为 Function 的参数执行，返回一个实现了 CompletionStage 接口的 CompletableFuture 对象实例。
thenCombine	有	a.thenCombine(b, Function) 返回一个新的CompletionStage，当 a,b 都正常完成时，将使用它两的结果作为参数执行 Function(结果a, 结果b)。 有关异常完成的规则，请参阅CompletionStage文档。
thenAccept	有	a.thenAccept(Consumer) 添加 Consumer 作为回调函数 ，a 执行完成后，结果传给 Consumer 执行。
thenAcceptBoth	有	a.thenAcceptBoth(BiConsumer) 添加 BiConsumer 作为回调函数 ，a, b 都执行完成后，结果传给 BiConsumer 执行。
whenComplete	有	a.whenComplete(BiConsumer) 返回一个新的CompletionStage，具有与此阶段相同的结果或异常，在此阶段完成时执行给定的操作。
exceptionally		a.exceptionally(Function) 如果异常返回给定值，否则正常返回。

acceptEither 谁先完成用谁

public class CompletableFutureTest {
    private String getName(long millis, String name) {
        try { Thread.sleep(millis); } catch (InterruptedException e) { throw new RuntimeException(e); }
        return name;
    }
    @Test
    public void acceptEitherTest() {
        CompletableFuture future1 = CompletableFuture.supplyAsync(()->{
            return getName(2000,"柳州螺蛳粉");
        });
        CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
            return getName(1000,"津市牛肉粉");
        });
        
        CompletableFuture completableFuture = future1.acceptEither(future2, System.out::println);
        completableFuture.join();
    }
}

津市牛肉粉

thenAccept 完成后消费结果

    @Test
    public void thenAcceptTest() {
        CompletableFuture future1 = CompletableFuture.supplyAsync(()->{
            return getName(2000,"柳州螺蛳粉");
        });
        future1.thenAccept(System.out::println).join();
    }

handle 完成后加工结果并返回

CompletableFuture handle = future1.handle((a, throwable) -> a + "加酸笋");
System.out.println(handle.get());

obtrudeValue 直接设置一个值为结果

@Test
public void obtrudeValueTest() throws ExecutionException, InterruptedException {
    CompletableFuture future1 = CompletableFuture.supplyAsync(()->{
        return getName(100000000,"柳州螺蛳粉");
    });
    future1.obtrudeValue("兰州拉面");
    System.out.println(future1.get());
}

thenAcceptBoth 两个都完成就消费

@Test
public void thenAcceptBothTest() throws ExecutionException, InterruptedException {
    CompletableFuture future1 = CompletableFuture.supplyAsync(()->{
        return getName(1000,"柳州螺蛳粉");
    });
    CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
        return getName(5000,"津市牛肉粉");
    });
    CompletableFuture completableFuture = future1.thenAcceptBoth(future2, (a, b) -> System.out.println(a + " - " + b));
    completableFuture.get();
}

whenComplete 完成后返回前，可读取结果

@Test
public void whenCompleteTest() throws ExecutionException, InterruptedException {
    CompletableFuture future1 = CompletableFuture.supplyAsync(()->{
        return getName(1000,"柳州螺蛳粉");
    });
    CompletableFuture stringCompletableFuture = future1.whenComplete((a, ex) -> {
        // 拿到异常 ex （如果有的话）
        if(ex instanceof Exception){ }
        // 拿到任务结果 result
        System.out.println("whenComplete 中取到的：" + result);
    });
    System.out.println(stringCompletableFuture.get());
}

exceptionally 如果异常返回给定值，否则正常返回

@Test
public void exceptionallyTest() throws ExecutionException, InterruptedException {
    CompletableFuture future1 = CompletableFuture.supplyAsync(()->{
        int i = 1/0;
        return getName(1000,"柳州螺蛳粉");
    });

    // 如果任务抛异常了，则返回给定值
    CompletableFuture stringCompletableFuture = future1.exceptionally((ex) ->{
        System.out.println(ex.getMessage());
        return "9527";
    });
    System.out.println("拿到结果：" + stringCompletableFuture.get());
}

参考资料

java.util.concurrent.CompletableFuture
java.util.concurrent.CompletionStage

java.util.concurrent.Future
java.util.concurrent.Executor
java.util.concurrent.Executors
java.util.concurrent.ExecutorService

Runtime.getRuntime().availableProcessors() 返回Java虚拟机可用的处理器数量