Java8实战-总结50

CompletableFuture：组合式异步编程

对多个异步任务进行流水线操作

对 Future 和 CompletableFuture 的回顾

CompletableFuture利用Lambda表达式以声明式的API提供了一种机制，能够用最有效的方式，非常容易地将多个以同步或异步方式执行复杂操作的任务结合到一起。为了更直观地感受一下使用CompletableFuture在代码可读性上带来的巨大提升，可以尝试仅使用Java 7中提供的特性，重新实现前面代码的功能。下面的代码展示了如何实现这一效果（利用Java 7的方法合并两个Future对象）：

ExecutorService executor = Executors.newCachedThreadPool(); //创建一个ExecutorService将任务提交到线程池

//创建一个查询欧元到美元转换汇率的Future
final Future<Double> futureRate = executor.submit(new Callable<Double>() {
	public Double call() { 
		return exchangeService.getRate(Money.EUR, Money.USD); 
	}}); 
	
Future<Double> futurePriceInUSD = executor.submit(new Callable<Double>() {
	public Double call() {
		double priceInEUR = shop.getPrice(product); //在第二个 Future中查询指定商店中特定商品的价格
		return priceInEUR * futureRate.get(); //在查找价格操作的同一个Future中，将价格和汇率做乘法计算出汇后价格
	}}); 

在上面的代码中，通过向执行器提交一个Callable对象的方式创建了第一个Future对象，向外部服务查询欧元和美元之间的转换汇率。紧接着，你创建了第二个Future对象，查询指定商店中特定商品的欧元价格。最终，用与之前代码一样的方式，在同一个Future中通过查询商店得到的欧元商品价格乘以汇率得到了最终的价格。注意，之前的代码中中如果使用thenCombineAsync，不使用thenCombine，像上面的代码一样，采用第三个Future单独进行商品价格和汇率的乘法运算，效果是几乎相同的。这两种实现看起来没太大区别，原因是你只对两个Future进行了合并。通过这两段代码，我们能看到创建流水线对同步和异步操作进行混合操作有多么简单，随着处理任务和需要合并结果数目的增加，这种声明式程序设计的优势也愈发明显。

你的“最佳价格查询器”应用基本已经完成，不过还缺失了一些元素。你会希望尽快将不同商店中的商品价格呈现给你的用户（这是车辆保险或者机票比价网站的典型需求），而不是像你之前那样，等所有的数据都完备之后再呈现。接下来的一节，你会了解如何通过响应CompletableFuture的completion事件实现这一功能（与此相反，“调用get或者join方法只会造成阻塞，直到CompletableFuture完成才能继续往下运行）。

响应 CompletableFuture 的 completion 事件

下面这部分的所有示例代码都是通过在响应之前添加1秒钟的等待延迟模拟方法的远程调用。毫无疑问，现实世界中，你的应用访问各个远程服务时很可能遭遇无法预知的延迟，触发的原因多种多样，从服务器的负荷到网络的延迟，有些甚至是源于远程服务如何评估你应用的商业价值，即可能相对于其他的应用，你的应用每次查询的消耗时间更长。

由于这些原因，你希望购买的商品在某些商店的查询速度要比另一些商店更快。以下面的代码清单为例，使用randomDelay方法取代原来的固定延迟。下面的代码是一个模拟生成0.5秒至2.5秒随机延迟的方法：

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

目前为止，你实现的findPrices方法只有在取得所有商店的返回值时才显示商品的价格。而你希望的效果是，只要有商店返回商品价格就在第一时间显示返回值，不再等待那些还未返回的商店（有些甚至会发生超时）。你如何实现这种更进一步的改进要求呢？

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

要避免的首要问题是，等待创建一个包含了所有价格的List创建完成。你应该做的是直接处理CompletableFuture流，这样每个CompletableFuture都在为某个商店执行必要的操作。为了实现这一目标，在下面的代码清单中，会对前面代码实现的第一部分进行重构，实现findPricesStream方法来生成一个由CompletableFuture构成的流。重构findPrices方法返回一个由Future构成的流：

public Stream<CompletableFuture<String>> findPricesStream(String product) {
	return shops.stream()
		.map(shop -> CompletableFuture.supplyAsync( 
				() -> shop.getPrice(product), executor))
		.map(future -> future.thenApply(Quote::parse))
		.map(future -> future.thenCompose(quote -> 
				CompletableFuture.supplyAsync(
					() -> Discount.applyDiscount(quote), executor))); 
} 

现在，你为findPricesStream方法返回的Stream添加了第四个map操作，在此之前，你已经在该方法内部调用了三次 map 。这个新添加的操作其实很简单，只是在每个CompletableFuture上注册一个操作，该操作会在CompletableFuture完成执行后使用它的返回值。Java 8的CompletableFuture通过thenAccept方法提供了这一功能，它接收CompletableFuture执行完毕后的返回值做参数。在这里的例子中，该值是由Discount服务返回的字符串值，它包含了提供请求商品的商店名称及折扣价格，你想要做的操作也很简单，只是将结果打印输出：

findPricesStream("myPhone").map(f -> f.thenAccept(System.out::println)); 

注意，和之前的thenCompose和thenCombine方法一样，thenAccept方法也提供了一个异步版本，名为thenAcceptAsync。异步版本的方法会对处理结果的消费者进行调度，从线程池中选择一个新的线程继续执行，不再由同一个线程完成CompletableFuture的所有任务。因为你想要避免不必要的上下文切换，更重要的是你希望避免在等待线程上浪费时间，尽快响应CompletableFuture的completion事件，所以这里没有采用异步版本。由 于thenAccept方法已经定义了如何处理CompletableFuture返回的结果，一旦CompletableFuture计算得到结果，它就返回一个CompletableFuture<Void>。所以，map操作返回的是一个Stream<CompletableFuture<Void>>。对这个<CompletableFuture<Void>>对象，你能做的事非常有限，只能等待其运行结束，不过这也是你所期望的。你还希望能给最慢的商店一些机会，让它有机会打印输出返回的价格。为了实现这一目的，你可以把构成Stream的所有CompletableFuture<Void>对象放到一个数组中，等待所有的任务执行完成，代码如下所示（响应CompletableFuture的completion事件）:

CompletableFuture[] futures = findPricesStream("myPhone")
	.map(f -> f.thenAccept(System.out::println))
	.toArray(size -> new CompletableFuture[size]); 
CompletableFuture.allOf(futures).join(); 

allOf工厂方法接收一个由CompletableFuture构成的数组，数组中的所有CompletableFuture对象执行完成之后，它返回一个CompletableFuture<Void>对象。这意味着，如果你需要等待最初Stream中的所有 CompletableFuture对象执行完毕，对 allOf方法返回的CompletableFuture执行join操作是个不错的主意。这个方法对“最佳价格查询器”应用也是有用的，因为你的用户可能会困惑是否后面还有一些价格没有返回，使用这个方法，你可以在执行完毕之后打印输出一条消息“All shops returned results or timed out”。

然而在另一些场景中，你可能希望只要CompletableFuture对象数组中有任何一个执行完毕就不再等待，比如，你正在查询两个汇率服务器，任何一个返回了结果都能满足你的需求。在这种情况下，你可以使用一个类似的工厂方法anyOf。该方法接收一个CompletableFuture对象构成的数组，返回由第一个执行完毕的CompletableFuture对象的返回值构成的CompletableFuture<Object>。