对产品实现汇率换算服务（将两个CompletableFuture对象整合起来，无论它们是否存在依赖）

需求

有一家商店提供的价格是以欧元（EUR）计价的，但是你希望以美元的方式提供给你的客户。你可以用异步的方式向商店查询指定商品的价格，同时从远程的汇率服务那里查到欧元和美元之间的汇率。当二者都结束时，再将这两个结果结合起来，用返回的商品价格乘以当时的汇率，得到以美元计价的商品价格。

实现源码

public class ExchangeService {public enum Money {USD(1.0), EUR(1.35387), GBP(1.69715), CAD(.92106), MXN(.07683);private final double rate;Money(double rate) {this.rate = rate;}}public static double getRate(Money source, Money destination) {return getRateWithDelay(source, destination);}private static double getRateWithDelay(Money source, Money destination) {Util.delay();return destination.rate / source.rate;}
}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"));private final Executor executor = Executors.newFixedThreadPool(shops.size(), ExecuterThreadFactoryBuilder.build());
}

实现方案

你对一个CompletableFuture对象调用了thenCompose方法，并向其传递了第二个CompletableFuture，而第二个CompletableFuture又需要使用第一个CompletableFuture的执行结果作为输入。但是，另一种比较常见的情况是，你需要将两个完全不相干的CompletableFuture对象的结果整合起来，而且你也不希望等到第一个任务完全结束才开始第二项任务。你应该使用thenCombine方法。thenCombine方法也提供有一个Async的版本。这里，如果使用thenCombineAsync会导致BiFunction中定义的合并操作被提交到线程池中，由另一个任务以异步的方式执行。

方案1：

public List<String> findPricesInUSD(String product) {List<CompletableFuture<Double>> priceFutures = new ArrayList<>();for (Shop shop : shops) {CompletableFuture<Double> futurePriceInUSD = CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor).thenCombine(CompletableFuture.supplyAsync(() -> ExchangeService.getRate(Money.EUR, Money.USD), executor), (price, rate) -> price * rate);priceFutures.add(futurePriceInUSD);}List<String> prices = priceFutures.stream().map(CompletableFuture::join).map(price -> " price is " + price).collect(Collectors.toList());return prices;
}

方案2：

public List<String> findPricesInUSD2(String product) {List<CompletableFuture<String>> priceFutures = new ArrayList<>();for (Shop shop : shops) {CompletableFuture<String> futurePriceInUSD = CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor).thenCombine(CompletableFuture.supplyAsync(() -> ExchangeService.getRate(Money.EUR, Money.USD), executor), (price, rate) -> price * rate).thenApply(price -> shop.getName() + " price is " + price);priceFutures.add(futurePriceInUSD);}List<String> prices = priceFutures.stream().map(CompletableFuture::join).collect(Collectors.toList());return prices;
}

方案3：

public List<String> findPricesInUSD3(String product) {List<CompletableFuture<String>> priceFutures = shops.stream().map(shop -> CompletableFuture.supplyAsync(() -> shop.getPrice(product)).thenCombine(CompletableFuture.supplyAsync(() -> ExchangeService.getRate(Money.EUR, Money.USD)), (price, rate) -> price * rate).thenApply(price -> shop.getName() + " price is " + price)).collect(Collectors.toList());List<String> prices = priceFutures.stream().map(CompletableFuture::join).collect(Collectors.toList());return prices;
}

方案4：使用Java 7中提供的特性完成上述功能

// 为了更直观地感受一下使用CompletableFuture在代码可读性上带来的巨大提升，尝试仅使用Java 7中提供的特性实现以下如上实现
public List<String> findPricesInUSDJava7(String product) {ExecutorService executor = Executors.newCachedThreadPool();List<Future<Double>> priceFutures = new ArrayList<>();for (Shop shop : shops) {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() {try {double priceInEUR = shop.getPrice(product);return priceInEUR * futureRate.get();} catch (InterruptedException | ExecutionException e) {throw new RuntimeException(e.getMessage(), e);}}});priceFutures.add(futurePriceInUSD);}List<String> prices = new ArrayList<>();for (Future<Double> priceFuture : priceFutures) {try {prices.add(" price is " + priceFuture.get());}catch (ExecutionException | InterruptedException e) {e.printStackTrace();}}return prices;
}

我们能看到创建流水线对同步和异步操作进行混合操作有多么简单，随着处理任务和需要合并结果数目的增加，这种声明式程序设计的优势也愈发明显。

测试

public static void main(String[] args) {StopWatch stopWatch = new StopWatch("性能比较");execute("combined USD findPricesInUSDJava7", () -> bestPriceFinder.findPricesInUSDJava7("myPhone27S"), stopWatch);execute("combined USD CompletableFuture", () -> bestPriceFinder.findPricesInUSD("myPhone27S"), stopWatch);execute("combined USD CompletableFuture v2", () -> bestPriceFinder.findPricesInUSD2("myPhone27S"), stopWatch);execute("combined USD CompletableFuture v3", () -> bestPriceFinder.findPricesInUSD3("myPhone27S"), stopWatch);StopWatchUtils.logStopWatch(stopWatch);
}
private static void execute(String msg, Supplier<List<String>> s, StopWatch stopWatch) {stopWatch.start(msg);System.out.println(s.get());stopWatch.stop();System.out.println();
}[ price is 91.040141702717,  price is 125.1710573102377,  price is 158.16078708139523,  price is 136.45612204497712,  price is 130.0591978746988][ price is 145.62246618545896,  price is 123.78887748261509,  price is 142.17561724598045,  price is 147.48700495707948,  price is 148.2901027867818][BestPrice price is 126.3823279607243, LetsSaveBig price is 124.52723804111048, MyFavoriteShop price is 129.1051274535831, BuyItAll price is 114.36072566615553, ShopEasy price is 121.11783256695436][BestPrice price is 168.06251816668828, LetsSaveBig price is 148.38498827435606, MyFavoriteShop price is 119.0272869408407, BuyItAll price is 115.15446874021768, ShopEasy price is 153.35355439427738]性能比较 total cost time = 6045 ms
combined USD findPricesInUSDJava7        : 1008 ms, 16.67%
combined USD CompletableFuture           : 2009 ms, 33.23%
combined USD CompletableFuture v2        : 2015 ms, 33.33%
combined USD CompletableFuture v3        : 1012 ms, 16.74%