CompletableFuture多线程并发异步编程

//创建ExecutorService，并且通过它向线程池提交任务
ExecutorService executorService = Executors. newCachedThreadPool();
//向线程池提交任务
Future future = executorService.submit( new Callable() {
@Override
public Double call() throws Exception {
return doSomeLongTime();
}
});
//当前线程继续做其他事情
doSomeElse();
//获取异步操作的结果
try {
double result = ( double)future.get( 1,TimeUnit. SECONDS);
} catch (InterruptedException e) {
//线程被中断
e.printStackTrace();
} catch (ExecutionException e) {
//线程抛出异常
e.printStackTrace();
} catch (TimeoutException e) {
//线程超时
e.printStackTrace();
}

public class Shop {
private String name;
private Random random = new Random();




public Shop(String name) {
this. name = name;
}




//直接获取价格
public double getPrice(String product){
return calculatePrice(product);
}
//模拟延迟
public static void delay(){
try {
Thread. sleep( 1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//模拟获取价格的服务
private double calculatePrice(String product){
delay();
return random.nextDouble() * product.charAt( 0) + product.charAt( 1);
}




//异步获取价格
public Future getPriceAsync(String product){
CompletableFuture future = new CompletableFuture<>();
new Thread(() -> {
double price = calculatePrice(product);
future.complete(price);
}).start();
return future;
}
}

public class Client {
public static void main(String[] args){
Shop shop = new Shop( "BestShop");
long start = System. currentTimeMillis();
Future future = shop.getPriceAsync( "My Favorite");
long invocatiOnTime= System. currentTimeMillis() - start;
System. out.println( "调用接口时间：" + invocationTime + "毫秒");




doSomethingElse();




try {
double price = future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}




long retrievalTime = System. currentTimeMillis() - start;
System. out.println( "返回价格消耗时间：" + retrievalTime + "毫秒");




}




public static void doSomethingElse(){
System. out.println( "做其他的事情。。。");
}
}

public static void main(String[] args){
long start = System. currentTimeMillis();
System. out.println( findPrice( "java8实战"));
long duration = System. currentTimeMillis() - start;
System. out.println( "总消耗时间：" + duration + "毫秒");
}




public static List findPrice(String product){
List shops = Arrays. asList( new Shop( "sunjin.org"),
new Shop( "加瓦匠"),
new Shop( "京东商城"),
new Shop( "天猫商城"));
return shops.parallelStream()
.map(shop -> String. format( "%s 的价格是 %.2f", shop.getName(),shop.getPrice(product)))
.collect(Collectors. toList());




}

public static List<String> findPrice2(String product){
List<CompletableFuture<String>> priceFuture = shops.stream()
.map(shop -> CompletableFuture. supplyAsync( // 使用异步的方式计算每种商品的价格
() -> shop.getName() + " 的价格是 " + shop.getPrice(product)))
.collect( toList());




return priceFuture.stream()
.map(CompletableFuture::join) //join 操作等待所有异步操作的结果
.collect( toList());
}
long start = System. currentTimeMillis();
System. out.println( findPrice2( "java8实战"));
long duration = System. currentTimeMillis() - start;
System.out.println("总消耗时间：" + duration + "毫秒");

[sunjin.org 的价格是 163.26300732704362, 加瓦匠 的价格是 172.3990021133357, 京东商城 的价格是 143.43284048828846, 天猫商城 的价格是 139.768245677111]
总消耗时间：2235毫秒

获取价格字符串
public String getPrice2(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);
}




解析价格字符串的报价类
public class Quote {
private final String shopName;
private final double price;
private final Discount.Code code;




public Quote(String shopName, double price, Discount.Code code) {
this. shopName = shopName;
this. price = price;
this. code = code;
}




public String getShopName() {
return shopName;
}




public double getPrice() {
return price;
}




public Discount.Code getCode() {
return code;
}




public static Quote parse(String s){
String[] arr = s.split( ":");
return new Quote(arr[ 0], Double. valueOf(arr[ 1]), Discount.Code. valueOf(arr[ 2]));
}
}

public class Discount {
public enum Code{
NONE( 0), SILVER( 5), 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.getCode());
}


public static double apply( double price, Code code){
delay();
return price * ( 100 - code. percentage) 100 ;
}




public static void delay(){
try {
Thread. sleep( 1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}




public static List findPrice3(String product){
return shops.stream()
.map(shop -> shop.getPrice2(product)) //获取原始报价
.map(Quote:: parse) //解析报价字符串
.map(Discount:: applyDiscount) //调用折扣服务应用报价折扣
.collect( toList());
}

public static List<String> findPrice4(String product){
List<CompletableFuture<String>> priceFuture = shops.stream()
.map(shop -> CompletableFuture. supplyAsync( // 异步获取价格
() -> shop.getPrice2(product), executor))
.map(future -> future.thenApply(Quote:: parse)) // 获取到价格后对价格解析
.map(future -> future.thenCompose(quote -> CompletableFuture. supplyAsync( // 另一个异步任务构造异步应用报价
() -> Discount. applyDiscount(quote), executor)))
.collect( toList());




return priceFuture.stream()
.map(CompletableFuture::join) //join 操作和get操作有相同的含义，等待所有异步操作的结果。
.collect( toList());
}

List> priceFuture = shops.stream()
.map(shop -> CompletableFuture. supplyAsync( // 异步获取价格
() -> shop.getPrice(product), executor))
.map(future -> future.thenCombine(CompletableFuture. supplyAsync( // 异步获取折扣率
() -> Discount. getRate(), executor)
, (price, rate) -> price * rate)) // 将两个异步任务的结果合并
.collect(toList());