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 对多个异步任务进行流水线操作(笔记) 的全部内容， 来源链接： utcz.com/z/390538.html
