网站建设维护更新,wordpress 邮件配置,wap版网站 加app提示,网站名称如何设置Java8实战-总结49 CompletableFuture#xff1a;组合式异步编程对多个异步任务进行流水线操作构造同步和异步操作将两个 CompletableFuture 对象整合起来#xff0c;无论它们是否存在依赖 CompletableFuture#xff1a;组合式异步编程
对多个异步任务进行流水线操作
构造同… Java8实战-总结49 CompletableFuture组合式异步编程对多个异步任务进行流水线操作构造同步和异步操作将两个 CompletableFuture 对象整合起来无论它们是否存在依赖 CompletableFuture组合式异步编程
对多个异步任务进行流水线操作
构造同步和异步操作
使用CompletableFuture提供的特性以异步方式重新实现findPrices方法。详细代码如下所示(使用CompletableFuture实现findPrices方法):
public ListString findPrices(String product) { ListCompletableFutureString priceFutures shops.stream().map(shop - CompletableFuture.supplyAsync( //以异步方式取得每个shop中指定产品的原始价格() - shop.getPrice(product), executor)).map(future - future.thenApply(Quote::parse)) //Quote对象存在时对其返回的值进行转换.map(future - future.thenCompose(quote - //使用另一个异步任务构造期望的Future申请折扣CompletableFuture.supplyAsync(() - Discount.applyDiscount(quote), executor))) .collect(toList()); return priceFutures.stream().map(CompletableFuture::join)//等待流中的所有Future执行完毕并提取各自的返回值.collect(toList());
} 这一次事情看起来变得更加复杂了这三次转换的流程如下图所示 进行的这三次map操作和前面代码中的同步方案没有太大的区别不过使用CompletableFuture类提供的特性在需要的地方把它们变成了异步操作。
获取价格 这三个操作中的第一个已经在各个例子中见过很多次只需要将Lambda表达式作为参数传递给supplyAsync工厂方法就可以以异步方式对shop进行查询。第一个转换的结果是一个StreamCompletableFutureString一旦运行结束每个CompletableFuture对象中都会包含对应shop返回的字符串。注意你对CompletableFuture进行了设置用前面代码中的方法向其传递了一个订制的执行器Executor。解析报价 现在需要进行第二次转换将字符串转变为订单。由于一般情况下解析操作不涉及任何远程服务也不会进行任何I/O操作它几乎可以在第一时间进行所以能够采用同步操作不会带来太多的延迟。由于这个原因你可以对第一步中生成的CompletableFuture对象调用它的thenApply将一个由字符串转换Quote的方法作为参数传递给它。注意到了吗直到调用的CompletableFuture执行结束使用的thenApply方法都不会阻塞代码的执行。这意味着CompletableFuture最终结束运行时你希望传递Lambda表达式给thenApply方法将Stream中的每个CompletableFutureString对象转换为对应的CompletableFutureQuote对象。你可以把这看成是为处理CompletableFuture的结果建立了一个菜单就像你曾经为Stream的流水线所做的事儿一样。为计算折扣价格构造Future 第三个map操作涉及联系远程的Discount服务为从商店中得到的原始价格申请折扣率。这一转换与前一个转换又不大一样因为这一转换需要远程执行或者就这个例子而言它需要模拟远程调用带来的延迟出于这一原因你也希望它能够异步执行。为了实现这一目标你像第一个调用传递getPrice给supplyAsync那样将这一操作以Lambda表达式的方式传递给了supplyAsync工厂方法该方法最终会返回另一个CompletableFuture对象。到目前为止你已经进行了两次异步操作用了两个不同的CompletableFutures对象进行建模你希望能把它们以级联的方式串接起来进行工作。从shop对象中获取价格接着把价格转换为Quote。拿到返回的Quote对象将其作为参数传递给Discount服务取得最终的折扣价格。
Java 8的 CompletableFuture API提供了名为thenCompose的方法它就是专门为这一目的而设计的thenCompose方法允许你对两个异步操作进行流水线第一个操作完成时将其结果作为参数传递给第二个操作。换句话说你可以创建两个CompletableFutures对象对第一个CompletableFuture对象调用 thenCompose并向其传递一个函数。当第一个CompletableFuture执行完毕后它的结果将作为该函数的参数这个函数的返回值是以第一个CompletableFuture的返回做输入计算出的第二个CompletableFuture对象。使用这种方式即使Future在向不同的商店收集报价主线程还是能继续执行其他重要的操作比如响应 事件。 将这三次map操作的返回的Stream元素收集到一个列表你就得到了一个ListCompletableFutureString等这些CompletableFuture对象最终执行完毕就可以像之前代码中那样利用join取得它们的返回值。代码实现的新版findPrices方法产生的输出如下
[BestPrice price is 110.93, LetsSaveBig price is 135.58, MyFavoriteShop price is 192.72, BuyItAll price is 184.74, ShopEasy price is 167.28]
Done in 2035 msecs 上面代码中使用的thenCompose方法像CompletableFuture类中的其他方法一样也提供了一个以Async后缀结尾的版本thenComposeAsync。通常而言名称中不带Async的方法和它的前一个任务一样在同一个线程中运行而名称以Async结尾的方法会将后续的任务提交到一个线程池所以每个任务是由不同的线程处理的。就这个例子而言第二个CompletableFuture对象的结果取决于第一个CompletableFuture所以无论你使用哪个版本的方法来处理CompletableFuture对象对于最终的结果或者大致的时间而言都没有多少差别。选择thenCompose方法的原因是因为它更高效一些因为少了很多线程切换的开销。
将两个 CompletableFuture 对象整合起来无论它们是否存在依赖
上面的代码中你对一个CompletableFuture对象调用了thenCompose方法并向其传递了第二个 CompletableFuture而第二个CompletableFuture又需要使用第一个CompletableFuture的执行结果作为输入。但是另一种比较常见的情况是你需要将两个完全不相干的CompletableFuture对象的结果整合起来而且你也不希望等到第一个任务完全结束才开始第二项任务。
这种情况你应该使用thenCombine方法它接收名为BiFunction的第二参数这个参数定义了当两个CompletableFuture对象完成计算后结果如何合并。同thenCompose方法一样thenCombine方法也提供有一个Async的版本。这里如果使用thenCombineAsync会导致BiFunction中定义的合并操作被提交到线程池中由另一个任务以异步的方式执行。回到我们正在运行的这个例子有一家商店提供的价格是以欧元EUR计价的但是你希望以美元的方式提供给你的客户。你可以用异步的方式向商店查询指定商品的价格同时从远程的汇率服务那里查到欧元和美元之间的汇率。当二者都结束时再将这两个结果结合起来用返回的商品价格乘以当时的汇率得到以美元计价的商品价格。用这种方式你需要使用第三个CompletableFuture 对象当前两个 CompletableFuture 计算出结果并由BiFunction方法完成合并后由它来最终结束这一任务代码清单如下所示
FutureDouble futurePriceInUSD CompletableFuture.supplyAsync(() - shop.getPrice(product)) //创建第一个任务查询商店取得商品的价格.thenCombine( CompletableFuture.supplyAsync( () - exchangeService.getRate(Money.EUR, Money.USD)), //创建第二个独立任务查询美元和欧元之间的转换汇率(price, rate) - price * rate);通过乘法 整合得到的商品价格和汇率); 这里整合的操作只是简单的乘法操作用另一个单独的任务对其进行操作有些浪费资源所以你只要使用thenCombine方法无需特别求助于异步版本的thenCombineAsync方法。下图展示了上面代码中创建的多个任务是如何在线程池中选择不同的线程执行的以及它们最终的运行结果又是如何整合的。
