首页 > 代码库 > Java8 对多个异步任务进行流水线操作(笔记)
Java8 对多个异步任务进行流水线操作(笔记)
现在我们要对商店商品进行折扣服务.每个折扣代码对应不同的折扣率,使用一个枚举变量Discount.Code来实现这一想法,具体代码如下所示. 以枚举类型定义的折扣代码 /** * 折扣服务api * * @author Darcy * Created by Administrator on 2017/3/17. */ public class Discount { public enum Code { NONE(0), SILVER(0), GOLD(10), PLATINUM(15), DIAMOND(20); private final int percentage; Code(int percentage) { this.percentage = percentage; } } public static String applyDiscount(Quote quote) { return quote.getShopName() + " price is " + Discount.apply(quote.getPrice(), quote.getDiscountCode()); } private static double apply(double price, Code code) { delay(); return price * (100 - code.percentage) / 100; } /** * 模拟计算,查询数据库等耗时 */ public static void delay() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } 修改商店返回价格的格式: public String getPrice(String product) { double price = calculatePrice(product); Discount.Code code = Discount.Code.values()[ random.nextInt(Discount.Code.values().length)]; return String.format("%s:%.2f:%s", name, price, code); } * 实现折扣服务 我们的商店已经能从不同的商店获得商品价格,解析结果字符串,针对每个字符串,查询折扣服务器取的折扣代码.这个流程决定了请求商品的最终折扣价格.我们将对商店返回的字符串的解析操作封装到了下面的Quote类中: /** * 商店返回消息实体,不可变对象模式 线程安全 * @author Darcy * Created by Administrator on 2017/3/17. */ public final class Quote { private final String shopName; private final double price; private final Discount.Code discountCode; public Quote(String shopName, double price, Discount.Code discountCode) { this.shopName = shopName; this.price = price; this.discountCode = 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); } public String getShopName() { return shopName; } public double getPrice() { return price; } public Discount.Code getDiscountCode() { return discountCode; } } Discount服务还提供了一个applyDiscount方法,它接收一个Quote对象,返回一个字符串,表示生成该Quote的shop中的折扣价格,代码如下: public static String applyDiscount(Quote quote) { return quote.getShopName() + " 商品原价: " + quote.getPrice() + " 折扣后价格: " + Discount.apply(quote.getPrice(), quote.getDiscountCode()); } private static double apply(double price, Code code) { delay(); return price * (100 - code.percentage) / 100; } * 使用Discount服务 /** * 商店折扣价格查询器 * * @param product 商品 * @return */ public static List<String> findprices(String product) { return shops .stream() .map(shop -> shop.getPrice(product)) .map(Quote::parse) .map(Discount::applyDiscount) .collect(Collectors.toList()); } 执行结果:
换成并行流: /** * 商店折扣价格查询器 * * @param product 商品 * @return */ public static List<String> findprices(String product) { return shops .parallelStream() .map(shop -> shop.getPrice(product)) .map(Quote::parse) .map(Discount::applyDiscount) .collect(Collectors.toList()); } 执行结果:
看到差距了吧 * 构建同步和异步操作 我们再次使用ComoletableFuture提供的特性.以异步方式重新实现findPrices方法,详细代码如下: /** * 商店折扣价格查询器(CompletableFuture方式) * * @param product 商品 * @return */ public static List<String> findPrices(String product) { List<CompletableFuture<String>> collect = shops .stream() //以异步凡是取得每个shop中指定产品的原始价格 .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor)) //Quote对象存在时,对其返回值进行转换 .map(future -> future.thenApply(Quote::parse)) //使用另一个异步任务构建期望的Future,申请折扣 thenCompose 将多个future组合 一个一个执行 .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor))) .collect(Collectors.toList()); return collect .stream() //等待流中所有的future执行完毕,并提取各自的返回值 .map(CompletableFuture::join) .collect(Collectors.toList()); } * 对最佳价格查询器应用的优化 上面的所有例子中都是通过响应之前添加1秒延迟的等待时间模拟方法的远程调用,毫无疑问,现实生活中,你的应用访问各个远程服务器时很可能遭遇无法预知的延迟,触发原因多种多样,从服务器的负荷到网络的延迟,有些甚至是源于远程服务如何评估你应用的商业价值, 由于这些原因,你希望购买的商品在某些原因的查询速度要比另一些商店更快,我们模拟了操作: /** * 模拟不同的商店 延迟不一样的情况 */ public static void randomDelay() { int delay = 500 + random.nextInt(2000); try { Thread.sleep(delay); } catch (InterruptedException e) { e.printStackTrace(); } } 重构findPrices方法 返回一个由Future构成的流: /** * 重构findPrices方法 返回一个由Future构成的流 * * @param product 商品 * @return */ public static Stream<CompletableFuture<String>> findProcesStream(String product) { return shops .stream() //以异步凡是取得每个shop中指定产品的原始价格 .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor)) //Quote对象存在时,对其返回值进行转换 .map(future -> future.thenApply(Quote::parse)) //使用另一个异步任务构建期望的Future,申请折扣 thenCompose 将多个future组合 一个一个执行 .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor))); } 付诸实现: long start = System.nanoTime(); CompletableFuture[] futures = findProcesStream("myPhones27s") //Java 8的CompletableFuture通过thenAccept方法 他接收CompletableFuture执行完毕的返回值作为参数. .map(f -> f.thenAccept( s -> System.out.println(s + " (done in " + ((System.nanoTime() - start) / 1_000_000) + " msecs)"))) .toArray(CompletableFuture[]::new); //allOf工厂方法接收一个由CompletableFuture构成的数组,数组中所有的CompletableFuture对象执行完毕后,它返回一个 //CompletableFuture<Void>对象,这意味着你需要等待最初Stream中所有的CompletableFuture对象执行完毕 //angOf该方法接收一个CompletableFuture对象构成的数组,返回由第一个执行完毕的CompletableFuture对象的返回值构成的CompletableFuture<Object> CompletableFuture.allOf(futures).join(); System.out.println("All shops have now responded in " + ((System.nanoTime() - start) / 1_000_000) + " msecs"); 执行结果:
Java8 对多个异步任务进行流水线操作(笔记)
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。