《Java8 实战》- 第十一章笔记（CompletableFuture：组合式异步编程）

CompletableFuture：组合式异步编程

最近这些年，两种趋势不断地推动我们反思我们设计软件的方式。第一种趋势和应用运行的硬件平台相关，第二种趋势与应用程序的架构相关，尤其是它们之间如何交互。我们在第 7 章中已经讨论过硬件平台的影响。我们注意到随着多核处理器的出现，提升应用程序处理速度最有效的方式是编写能充分发挥多核能力的软件。你已经看到通过切分大型的任务，让每个子任务并行运行，这一目标是能够实现的；你也已经了解相对直接使用线程的方式，使用分支/合并框架（在 Java 7 中引入）和并行流（在 Java 8 中新引入）能以更简单、更有效的方式实现这一目标。

第二种趋势反映在公共 API 日益增长的互联网服务应用。著名的互联网大鳄们纷纷提供了自己的公共 API 服务，比如谷歌提供了地理信息服务，Facebook 提供了社交信息服务，Twitter 提供了新闻服务。现在，很少有网站或者网络应用会以完全隔离的方式工作。更多的时候，我们看到的下一代网络应用都采用“混聚”（mash-up）的方式：它会使用来自多个来源的内容，将这些内容聚合在一起，方便用户的生活。

比如，你可能希望为你的法国客户提供指定主题的热点报道。为实现这一功能，你需要向谷歌或者 Twitter 的 API 请求所有语言中针对该主题最热门的评论，可能还需要依据你的内部算法对它们的相关性进行排序。之后，你可能还需要使用谷歌的翻译服务把它们翻译成法语，甚至利用谷歌地图服务定位出评论作者的位置信息，最终将所有这些信息聚集起来，呈现在你的网站上。

当然，如果某些外部网络服务发生响应慢的情况，你希望依旧能为用户提供部分信息，比如提供带问号标记的通用地图，以文本的方式显示信息，而不是呆呆地显示一片空白屏幕，直到地图服务器返回结果或者超时退出。

要实现类似的服务，你需要与互联网上的多个 Web 服务通信。可是，你并不希望因为等待某些服务的响应，阻塞应用程序的运行，浪费数十亿宝贵的 CPU 时钟周期。比如，不要因为等待 Facebook 的数据，暂停对来自 Twitter 的数据处理。

这些场景体现了多任务程序设计的另一面。第 7 章中介绍的分支/合并框架以及并行流是实现并行处理的宝贵工具；它们将一个操作切分为多个子操作，在多个不同的核、CPU 甚至是机器上并行地执行这些子操作。

与此相反，如果你的意图是实现并发，而非并行，或者你的主要目标是在同一个 CPU 上执行几个松耦合的任务，充分利用 CPU 的核，让其足够忙碌，从而最大化程序的吞吐量，那么你其实真正想做的是避免因为等待远程服务的返回，或者对数据库的查询，而阻塞线程的执行，浪费宝贵的计算资源，因为这种等待的时间很可能相当长。通过本章中你会了解，Future 接口，尤其是它的新版实现 CompletableFuture，是处理这种情况的利器。

Future 接口

Future 接口在 Java 5 中被引入，设计初衷是对将来某个时刻会发生的结果进行建模。它建模了一种异步计算，返回一个执行运算结果的引用，当运算结束后，这个引用被返回给调用方。在 Future 中触发那些潜在耗时的操作把调用线程解放出来，让它能继续执行其他有价值的工作，不再需要呆呆等待耗时的操作完成。打个比方，你可以把它想象成这样的场景：你拿了一袋子衣服到你中意的干洗店去洗。干洗店的员工会给你张发票，告诉你什么时候你的衣服会洗好（这就是一个 Future 事件）。衣服干洗的同时，你可以去做其他的事情。Future 的另一个优点是它比更底层的 Thread 更易用。要使用 Future，通常你只需要将耗时的操作封装在一个 Callable 对象中，再将它提交给 ExecutorService，就万事大吉了。下面这段代码展示了 Java 8 之前使用 Future 的一个例子。

ExecutorService executor = Executors.newCachedThreadPool();
Future<Double> future = executor.submit(new Callable<Double>() {
    public Double call() {
        return doSomeLongComputation();
    }
});
// 异步操作进行的同时，你可以做其他的事情
doSomethingElse();

try {
    Double result = future.get(1, TimeUnit.SECONDS);
} catch (ExecutionException ee) {
    // 计算抛出一个异常
} catch (InterruptedException ie) {
    // 当前线程在等待过程中被中断
} catch (TimeoutException te) {
    // 在Future对象完成之前超过已过期
}

这种编程方式让你的线程可以在 ExecutorService 以并发方式调用另一个线程执行耗时操作的同时，去执行一些其他的任务。接着，如果你已经运行到没有异步操作的结果就无法继续任何有意义的工作时，可以调用它的 get 方法去获取操作的结果。如果操作已经完成，该方法会立刻返回操作的结果，否则它会阻塞你的线程，直到操作完成，返回相应的结果。

你能想象这种场景存在怎样的问题吗？如果该长时间运行的操作永远不返回了会怎样？为了处理这种可能性，虽然 Future 提供了一个无需任何参数的 get 方法，我们还是推荐大家使用重载版本的 get 方法，它接受一个超时的参数，通过它，你可以定义你的线程等待 Future 结果的最长时间，而不是样永无止境地等待下去。

Future 接口的局限性

通过第一个例子，我们知道 Future 接口提供了方法来检测异步计算是否已经结束（使用 isDone 方法），等待异步操作结束，以及获取计算的结果。但是这些特性还不足以让你编写简洁的并发代码。比如，我们很难表述 Future 结果之间的依赖性；从文字描述上这很简单，“当长时间计算任务完成时，请将该计算的结果通知到另一个长时间运行的计算任务，这两个计算任务都完成后，将计算的结果与另一个查询操作结果合并”。但是，使用 Future 中提供的方法完成这样的操作又是另外一回事。这也是我们需要更具描述能力的特性的原因，比如下面这些。

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

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

使用 CompletableFuture 构建异步应用

为了展示 CompletableFuture 的强大特性，我们会创建一个名为“最佳价格查询器”（best-price-finder）的应用，它会查询多个在线商店，依据给定的产品或服务找出最低的价格。这个过程中，你会学到几个重要的技能。

• 首先，你会学到如何为你的客户提供异步 API（如果你拥有一间在线商店的话，这是非常有帮助的）。
• 其次，你会掌握如何让你使用了同步 API 的代码变为非阻塞代码。你会了解如何使用流水线将两个接续的异步操作合并为一个异步计算操作。这种情况肯定会出现，比如，在线商店返回了你想要购买商品的原始价格，并附带着一个折扣代码——最终，要计算出该商品的实际价格，你不得不访问第二个远程折扣服务，查询该折扣代码对应的折扣比率。
• 你还会学到如何以响应式的方式处理异步操作的完成事件，以及随着各个商店返回它的商品价格，最佳价格查询器如何持续地更新每种商品的最佳推荐，而不是等待所有的商店都返回他们各自的价格（这种方式存在着一定的风险，一旦某家商店的服务中断，用户可能遭遇白屏）。

实现异步 API

为了实现最佳价格查询器应用，让我们从每个商店都应该提供的 API 定义入手。首先，商店应该声明依据指定产品名称返回价格的方法：

public double getPrice(String product) {
    // 待实现
}

该方法的内部实现会查询商店的数据库，但也有可能执行一些其他耗时的任务，比如联系其他外部服务（比如，商店的供应商，或者跟制造商相关的推广折扣）。我们在本章剩下的内容中，采用 delay 方法模拟这些长期运行的方法的执行，它会人为地引入 1 秒钟的延迟，方法声明如下。

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

为了介绍本章的内容，getPrice 方法会调用 delay 方法，并返回一个随机计算的值，代码清单如下所示。返回随机计算的价格这段代码看起来有些取巧。它使用 charAt，依据产品的名称，生成一个随机值作为价格。

public class Shop {
    private final String name;
    private final Random random;

    public Shop(String name) {
        this.name = name;
        random = new Random(name.charAt(0) * name.charAt(1) * name.charAt(2));
    }

    public double getPrice(String product) {
        return calculatePrice(product);
    }

    private double calculatePrice(String product) {
        delay();
        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }
}

很明显，这个 API 的使用者（这个例子中为最佳价格查询器）调用该方法时，它依旧会被阻塞。为等待同步事件完成而等待 1 秒钟，这是无法接受的，尤其是考虑到最佳价格查询器对网络中的所有商店都要重复这种操作。本章接下来的小节中，你会了解如何以异步方式使用同步 API 解决这个问题。但是，出于学习如何设计异步 API 的考虑，我们会继续这一节的内容，假装我们还在深受这一困难的烦扰：你是一个睿智的商店店主，你已经意识到了这种同步 API 会为你的用户带来多么痛苦的体验，你希望以异步 API 的方式重写这段代码，让用户更流畅地访问你的网站。

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

为了实现这个目标，你首先需要将 getPrice 转换为 getPriceAsync 方法，并修改它的返回值：

public class Shop {
    ...
    public Future<Double> getPriceAsync(String product) {
        CompletableFuture<Double> futurePrice = new CompletableFuture<>();
        new Thread(() -> {
            double price = calculatePrice(product);
            futurePrice.complete(price);
        }).start();
        return futurePrice;
    }
    ...
}

在这段代码中，你创建了一个代表异步计算的 CompletableFuture 对象实例，它在计算完成时会包含计算的结果。接着，你调用 fork 创建了另一个线程去执行实际的价格计算工作，不等该耗时计算任务结束，直接返回一个 Future 实例。当请求的产品价格最终计算得出时，你可以使用它的 complete 方法，结束 completableFuture 对象的运行，并设置变量的值。很显然，这个新版 Future 的名称也解释了它所具有的特性。使用这个 API 的客户端，可以通过下面的这段代码对其进行调用。

public class ShopMain {

    public static void main(String[] args) {
        Shop shop = new Shop("最好的商店");
        long start = System.nanoTime();
        Future<Double> futurePrice = shop.getPriceAsync("我最喜欢的商品");
        long invocationTime = ((System.nanoTime() - start) / 1_000_000);
        System.out.println("调用时间 " + invocationTime);
        // 这里可以做其他的事情，比如查询其他的商店
        doSomethingElse();
        // 计算商品价格
        try {
            double price = futurePrice.get();
            System.out.printf("价格是 %.2f%n", price);
        } catch (ExecutionException | InterruptedException e) {
            throw new RuntimeException(e);
        }
        long retrievalTime = ((System.nanoTime() - start) / 1_000_000);
        System.out.println("计算价格时间 " + retrievalTime);
    }

    private static void doSomethingElse() {
        System.out.println("正在查询其他的商店...");
    }
}

我们看到这段代码中，客户向商店查询了某种商品的价格。由于商店提供了异步 API，该次调用立刻返回了一个 Future 对象，通过该对象客户可以在将来的某个时刻取得商品的价格。这种方式下，客户在进行商品价格查询的同时，还能执行一些其他的任务，比如查询其他家商店中商品的价格，不会呆呆地阻塞在那里等待第一家商店返回请求的结果。最后，如果所有有意义的工作都已经完成，客户所有要执行的工作都依赖于商品价格时，再调用 Future 的 get 方法。执行了这个操作后，客户要么获得 Future 中封装的值（如果异步任务已经完成），要么发生阻塞，直到该异步任务完成，期望的值能够访问。上面的代码中，输出的结果：

调用时间 116
正在查询其他的商店...
价格是 49107.07
计算价格时间 1172

你一定已经发现 getPriceAsync 方法的调用时间远远早于最终价格计算完成的时间，在之前的代码，你还会知道我们有可能避免发生客户端被阻塞的风险。实际上这非常简单，Future 执行完毕可以发送一个通知，仅在计算结果可用时执行一个由 Lambda 表达式或者方法引用定义的回调函数。不过，我们当下不会对此进行讨论，现在我们要解决的是另一个问题：如何正确地管理异步任务执行过程中可能出现的错误。

错误处理

如果没有意外，我们目前开发的代码工作得很正常。但是，如果价格计算过程中产生了错误会怎样呢？非常不幸，这种情况下你会得到一个相当糟糕的结果：用于提示错误的异常会被限制在试图计算商品价格的当前线程的范围内，最终会杀死该线程，而这会导致等待 get 方法返回结果的客户端永久地被阻塞。

客户端可以使用重载版本的 get 方法，它使用一个超时参数来避免发生这样的情况。这是一种值得推荐的做法，你应该尽量在你的代码中添加超时判断的逻辑，避免发生类似的问题。使用这种方法至少能防止程序永久地等待下去，超时发生时，程序会得到通知发生了 TimeoutException。不过，也因为如此，你不会有机会发现计算商品价格的线程内到底发生了什么问题才引发了这样的失效。为了让客户端能了解商店无法提供请求商品价格的原因，你需要使用 CompletableFuture 的 completeExceptionally 方法将导致 CompletableFuture 内发生问题的异常抛出。对代码优化后的结果如下所示。

public Future<Double> getPriceAsync(String product) {
    CompletableFuture<Double> futurePrice = new CompletableFuture<>();
    new Thread(() -> {
        try {
            double price = calculatePrice(product);
            // 如果价格计算正常结束，完成Future操作并设置商品价格
            futurePrice.complete(price);
        } catch (Exception e) {
            // 否则就抛出导致失败的异常，完成这次Future操作
            futurePrice.completeExceptionally(e);
        }
    }).start();
    return futurePrice;
}

客户端现在会收到一个 ExecutionException 异常，该异常接收了一个包含失败原因的 Exception 参数，即价格计算方法最初抛出的异常。所以，举例来说，如果该方法抛出了一个运行时异常“product not available”，客户端就会得到像下面这样一段 ExecutionException：

java.util.concurrent.ExecutionException: java.lang.RuntimeException: product
    not availableat java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2237)
    at xin.codedream.java8.chap11.AsyncShopClient.main(AsyncShopClient.java:14)
    ... 5 more
Caused by: java.lang.RuntimeException: product not available
    at xin.codedream.java8.chap11.AsyncShop.calculatePrice(AsyncShop.java:36)
    atxin.codedream.java8.chap11.AsyncShop.lambda$getPrice$0(AsyncShop.java:23)
    at xin.codedream.java8.chap11.AsyncShop$$Lambda$1/24071475.run(Unknown Source)
    at java.lang.Thread.run(Thread.java:744)

使用工厂方法 supplyAsync 创建 CompletableFuture
目前为止我们已经了解了如何通过编程创建 CompletableFuture 对象以及如何获取返回值，虽然看起来这些操作已经比较方便，但还有进一步提升的空间，CompletableFuture 类自身提供了大量精巧的工厂方法，使用这些方法能更容易地完成整个流程，还不用担心实现的细节。比如，采用 supplyAsync 方法后，你可以用一行语句重写 getPriceAsync 方法，如下所示。

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

太棒了！七八行才能实现的功能，我们现在只需要一行就可以搞定了！supplyAsync 方法接受一个生产者（Supplier）作为参数，返回一个 CompletableFuture 对象，该对象完成异步执行后会读取调用生产者方法的返回值。生产者方法会交由 ForkJoinPool 池中的某个执行线程（Executor）运行，但是你也可以使用 supplyAsync 方法的重载版本，传递第二个参数指定不同的执行线程执行生产者方法。一般而言，向 CompletableFuture 的工厂方法传递可选参数，指定生产者方法的执行线程是可行的，在后面，你会使用这一能力，后面我们将使用适合你应用特性的执行线程改善程序的性能。

接下来剩余部分中，我们会假设你非常不幸，无法控制 Shop 类提供 API 的具体实现，最终提供给你的 API 都是同步阻塞式的方法。这也是当你试图使用服务提供的 HTTP API 时最常发生的情况。你会学到如何以异步的方式查询多个商店，避免被单一的请求所阻塞，并由此提升你的“最佳价格查询器”的性能和吞吐量。

让你的代码免受阻塞之苦

所以，你已经被要求进行“最佳价格查询器”应用的开发了，不过你需要查询的所有商店都如上面开始时介绍的那样，只提供了同步 API。换句话说，你有一个商家的列表，如下所示：

public class BestPriceFinder {
    private final List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavoriteShop"),
            new Shop("BuyItAll"));
    ...
}

你需要使用下面这样的签名实现一个方法，它接受产品名作为参数，返回一个字符串列表，这个字符串列表中包括商店的名称、该商店中指定商品的价格：

public List<String> findPrices(String product);

你的第一个想法可能是使用我们在前面的章节中学习的 Stream 特性。你可能试图写出类似下面这个代码（是的，作为第一个方案，如果你想到这些已经相当棒了！）。

好吧，这段代码看起来非常直白。现在试着用该方法去查询你最近这些天疯狂着迷的唯一产品（是的，你已经猜到了，它就是 Old-Mi-Mix3）。此外，也请记录下方法的执行时间，通过这些数据，我们可以比较优化之后的方法会带来多大的性能提升，具体的代码如下。

public class BestPriceFinder {
    private final List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavoriteShop"),
            new Shop("BuyItAll"));

    public static void main(String[] args) {
        BestPriceFinder finder = new BestPriceFinder();
        finder.testFindPrices();
    }

    public void testFindPrices() {
        long start = System.nanoTime();
        System.out.println(findPrices("Old-Mi-Mix3"));
        long duration = (System.nanoTime() - start) / 1_000_000;
        System.out.println("完成时间 " + duration);
    }

    public List<String> findPrices(String product) {
        return shops.stream()
                .map(shop -> String.format("%s 价格 %.2f",
                        shop.getName(), shop.getPrice(product)))
                .collect(toList());
    }
}

输出结果：

[BestPrice 价格 109.64, LetsSaveBig 价格 143.13, MyFavoriteShop 价格 175.50, BuyItAll 价格 154.20]
完成时间 4184

正如你预期的，findPrices 方法的执行时间仅比 4 秒钟多了那么几百毫秒，因为对这 4 个商店的查询是顺序进行的，并且一个查询操作会阻塞另一个，每一个操作都要花费大约 1 秒左右的时间计算请求商品的价格。你怎样才能改进这个结果呢？

使用并行流对请求进行并行操作

如果你看了第七章的笔记，那么你应该想到的第一个，可能也是最快的改善方法是使用并行流来避免顺序计算，如下所示。

public List<String> findPricesParallel(String product) {
    return shops.parallelStream()
            .map(shop -> String.format("%s 价格 %.2f",
                    shop.getName(), shop.getPrice(product)))
            .collect(toList());
}

运行代码，与最初的代码执行结果相比较，你发现了新版 findPrices 的改进了吧。

[BestPrice 价格 109.64, LetsSaveBig 价格 143.13, MyFavoriteShop 价格 175.50, BuyItAll 价格 154.20]
完成时间 1248

相当不错啊！看起来这是个简单但有效的主意：现在对四个不同商店的查询实现了并行，所以完成所有操作的总耗时只有 1 秒多一点儿。你能做得更好吗？让我们尝试使用刚学过的 CompletableFuture，将 findPrices 方法中对不同商店的同步调用替换为异步调用。

使用 CompletableFuture 发起异步请求

你已经知道我们可以使用工厂方法 supplyAsync 创建 CompletableFuture 对象。让我们把它利用起来：

public List<CompletableFuture<String>> findPricesFuture(String product) {
    return shops.stream()
            .map(shop -> CompletableFuture.supplyAsync(() -> String.format("%s 价格 %.2f",
                    shop.getName(), shop.getPrice(product))))
            .collect(toList());
}

使用这种方式，你会得到一个 List<CompletableFuture>，列表中的每个 CompletableFuture 对象在计算完成后都包含商店的 String 类型的名称。但是，由于你用 CompletableFutures 实现的 findPrices 方法要求返回一个 List，你需要等待所有的 future 执行完毕，将其包含的值抽取出来，填充到列表中才能返回。

为了实现这个效果，你可以向最初的 List<CompletableFuture> 施加第二个 map 操作，对 List 中的所有 future 对象执行 join 操作，一个接一个地等待它们运行结束。注意 CompletableFuture 类中的 join 方法和 Future 接口中的 get 有相同的含义，并且也声明在 Future 接口中，它们唯一的不同是 join 不会抛出任何检测到的异常。使用它你不再需要使用 try/catch 语句块让你传递给第二个 map 方法的 Lambda 表达式变得过于臃肿。所有这些整合在一起，你就可以重新实现 findPrices 了，具体代码如下。

public List<String> findPrices(String product) {
    List<CompletableFuture<String>> priceFutures = shops.stream()
            .map(shop -> CompletableFuture.supplyAsync(() -> String.format("%s 价格 %.2f",
                    shop.getName(), shop.getPrice(product))))
            .collect(toList());

    return priceFutures.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());
}

运行下代码了解下第三个版本 findPrices 方法的性能，你会得到下面这几行输出：

[BestPrice 价格 109.64, LetsSaveBig 价格 143.13, MyFavoriteShop 价格 175.50, BuyItAll 价格 154.20]
完成时间 2207

这个结果让人相当失望，不是吗？超过 2 秒意味着利用 CompletableFuture 实现的版本，比刚开始的代码中的原生顺序执行且会发生阻塞的版本快。但是它的用时也差不多是使用并行流的前一个版本的两倍。尤其是，考虑到从顺序执行的版本转换到并行流的版本只做了非常小的改动，就让人更加沮丧。

与此形成鲜明对比的是，我们为采用 CompletableFutures 完成的新版方法做了大量的工作！但，这就是全部的真相吗？这种场景下使用 CompletableFutures 真的是浪费时间吗？或者我们可能漏掉了某些重要的东西？继续往下探究之前，让我们休息几分钟，尤其是想想你测试代码的机器是否足以以并行方式运行四个线程。

寻找更好的方案

并行流的版本工作得非常好，那是因为它能并行地执行四个任务，所以它几乎能为每个商家分配一个线程。但是，如果你想要增加第五个商家到商店列表中，让你的“最佳价格查询”应用对其进行处理，这时会发生什么情况？

public class BestPriceFinder {
    private final List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavoriteShop"),
            new Shop("BuyItAll"),
            new Shop("ShopEasy"));
    ...

    public List<String> findPricesParallel(String product) {
        return shops.parallelStream()
                .map(shop -> String.format("%s 价格 %.2f",
                        shop.getName(), shop.getPrice(product)))
                .collect(toList());
    }

    public List<String> findPricesSequential(String product) {
        return shops.stream()
                .map(shop -> String.format("%s 价格 %.2f",
                        shop.getName(), shop.getPrice(product)))
                .collect(toList());
    }


    public List<String> findPricesFuture(String product) {
        List<CompletableFuture<String>> priceFutures = shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() -> String.format("%s 价格 %.2f",
                        shop.getName(), shop.getPrice(product))))
                .collect(toList());

        return priceFutures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
    }
}

public class BestPriceFinderMain {

    private static BestPriceFinder bestPriceFinder = new BestPriceFinder();

    public static void main(String[] args) {
        execute("sequential", () -> bestPriceFinder.findPricesSequential("Old-Mi-Mix3"));
    }

    private static void execute(String msg, Supplier<List<String>> s) {
        long start = System.nanoTime();
        System.out.println(s.get());
        long duration = (System.nanoTime() - start) / 1_000_000;
        System.out.println(msg + " 完成时间 " + duration);
    }
}

毫不意外，顺序执行版本的执行还是需要大约 5 秒多钟的时间，下面是执行的输出：

[BestPrice 价格 109.64, LetsSaveBig 价格 143.13, MyFavoriteShop 价格 175.50, BuyItAll 价格 154.20, ShopEasy 价格 147.92]
sequential 完成时间 5139

非常不幸，并行流版本的程序这次比之前也多消耗了差不多 1 秒钟的时间，因为可以并行运行（通用线程池中处于可用状态的）的四个线程现在都处于繁忙状态，都在对前 4 个商店进行查询。第五个查询只能等到前面某一个操作完成释放出空闲线程才能继续，它的运行结果如下：

[BestPrice 价格 163.19, LetsSaveBig 价格 141.77, MyFavoriteShop 价格 159.81, BuyItAll 价格 165.02, ShopEasy 价格 165.81]
parallel 完成时间 2106

CompletableFuture 版本的程序结果如何呢？我们也试着添加第 5 个商店对其进行了测试，结果如下：

[BestPrice 价格 144.31, LetsSaveBig 价格 142.49, MyFavoriteShop 价格 146.99, BuyItAll 价格 132.52, ShopEasy 价格 139.15]
composed CompletableFuture 完成时间 2004

CompletableFuture 版本的程序似乎比并行流版本的程序还快那么一点儿。但是最后这个版本也不太令人满意。比如，如果你试图让你的代码处理 9 个商店，并行流版本耗时 3143 毫秒，而 CompletableFuture 版本耗时 3009 毫秒。它们看起来不相伯仲，究其原因都一样：它们内部采用的是同样的通用线程池，默认都使用固定数目的线程，具体线程数取决于 Runtime.getRuntime().availableProcessors()的返回值。然而，CompletableFuture 具有一定的优势，因为它允许你对执行器（Executor）进行配置，尤其是线程池的大小，让它以更适合应用需求的方式进行配置，满足程序的要求，而这是并行流 API 无法提供的。让我们看看你怎样利用这种配置上的灵活性带来实际应用程序性能上的提升。

使用定制的执行器

就这个主题而言，明智的选择似乎是创建一个配有线程池的执行器，线程池中线程的数目取决于你预计你的应用需要处理的负荷，但是你该如何选择合适的线程数目呢？

你的应用 99% 的时间都在等待商店的响应，所以估算出的 W/C 比率为 100。这意味着如果你期望的 CPU 利用率是 100%，你需要创建一个拥有 400 个线程的线程池。实际操作中，如果你创建的线程数比商店的数目更多，反而是一种浪费，因为这样做之后，你线程池中的有些线程根本没有机会被使用。出于这种考虑，我们建议你将执行器使用的线程数，与你需要查询的商店数目设定为同一个值，这样每个商店都应该对应一个服务线程。不过，为了避免发生由于商店的数目过多导致服务器超负荷而崩溃，你还是需要设置一个上限，比如 100 个线程。代码清单如下所示。

private final Executor executor = Executors.newFixedThreadPool(100, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setDaemon(true);
            return t;
        }
    });

注意，你现在正创建的是一个由守护线程构成的线程池。Java 程序无法终止或者退出一个正在运行中的线程，所以最后剩下的那个线程会由于一直等待无法发生的事件而引发问题。与此相反，如果将线程标记为守护进程，意味着程序退出时它也会被回收。这二者之间没有性能上的差异。现在，你可以将执行器作为第二个参数传递给 supplyAsync 工厂方法了。比如，你现在可以按照下面的方式创建一个可查询指定商品价格的 CompletableFuture 对象：

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

改进之后，使用 CompletableFuture 方案的程序处理 5 个商店结果：

[BestPrice 价格 144.31, LetsSaveBig 价格 142.49, MyFavoriteShop 价格 146.99, BuyItAll 价格 132.52, ShopEasy 价格 139.15]
composed CompletableFuture 完成时间 1004

这个例子证明了要创建更适合你的应用特性的执行器，利用 CompletableFutures 向其提交任务执行是个不错的主意。处理需大量使用异步操作的情况时，这几乎是最有效的策略。

并行——使用流还是 CompletableFutures？

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

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

现在你已经了解了如何利用 CompletableFuture 为你的用户提供异步 API，以及如何将一个同步又缓慢的服务转换为异步的服务。不过到目前为止，我们每个 Future 中进行的都是单次的操作。

代码

Github: chap11

Gitee: chap11