架设我是一个店主,提供了商品价格查询api 传过来商品名字,我就返回价格
public double getPrice(String product) {
//一些耗时的操作 比如查询数据库
return calculatePrice(product);
}
CompletableFuture
明显 这是一个阻塞式的调用,可以改进为异步的↓
public Future<Double> getPriceAsync(String product) {
CompletableFuture<Double> futurePrice = new CompletableFuture<>();
Thread thread = new Thread(() -> {
double price = calculatePrice(product);
//结果可以通过futurePrice来get
futurePrice.complete(price);
});
thread.start();
return futurePrice;
}
不直接返回结果 而是一个Future对象
这样客户就可以这样调用
Future<Double> futurePrice = shop.getPriceAsync("my favorite product");
//做写其他的事,直到必须用到价格
double price = futurePrice.get(1,TimeUnit.SECONDS);//等待一秒,必须拿到价格
completeExceptionally
如果我这边查询价格的方法出了错,希望把异常报给客户,改进
public Future<Double> getPrice(String product) {
CompletableFuture<Double> futurePrice = new CompletableFuture<>();
new Thread( () -> {
try {
double price = calculatePrice(product);
futurePrice.complete(price);
} catch (Exception ex) {
//这句不同,放入异常,抛给调用方
futurePrice.completeExceptionally(ex);
}
}).start();
return futurePrice;
}
CompletableFuture.supplyAsync
我发现CompletableFuture本身提供了工厂方法包装了上面这坨,功能一样的简短版改进↓
public Future<Double> getPrice(String product) {
return CompletableFuture.supplyAsync(() -> calculatePrice(product));
}
调用方改进
架设我不是店主,所以getPrice没得改进,还是第一个阻塞的版本
我要查这些店list
private final List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
new Shop("LetsSaveBig"),
new Shop("MyFavoriteShop"),
new Shop("BuyItAll"),
new Shop("ShopEasy"));
parallelStream
适合计算密集型,都是计算不用怎么等
//并行流 几个核就几个线程
public List<String> findPricesParallel(String product) {
return shops.parallelStream()
.map(shop -> shop.getName() + " price is " + shop.getPrice(product))
.collect(Collectors.toList());
}
Executor 定制
自定义线程数
但是发现,大量时间都在等待商店响应,而不是计算,线程数超过cpu核数更加高效,为了自定义线程数,用自制的执行器,改用CompletableFuture
定制执行器
ThreadFactory threadFactory=new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setDaemon(true);
return t;
}
};
//指定线程池大小 为店的个数
private final Executor executor = Executors.newFixedThreadPool(shops.size(),threadFactory);
//CompletableFuture 先拿回 List<CompletableFuture<String>> 再阻塞保证拿到每个String 达成返回 List<String>
//定制了线程池
public List<String> findPricesFuture(String product) {
List<CompletableFuture<String>> priceFutures =
shops.stream()
.map(shop -> {
//耗时任务
Supplier<String> stringSupplier = () -> shop.getName() + " price is "
+ shop.getPrice(product);
//除了任务之外 多一个执行器入参
return CompletableFuture.supplyAsync(stringSupplier, executor);
})
.collect(Collectors.toList());
List<String> prices = priceFutures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList());
return prices;
}
thenCompose
第一个CompletableFuture作为第二个CompletableFuture的输入
现在有个新需求,商店不止会返回价格,还会返回折扣,我拿着折扣号和价格,再去折扣中心计算一下返回的才是最后实际价格
public Stream<CompletableFuture<String>> findPricesStream(String product) {
return shops.stream()
.map(shop -> CompletableFuture.supplyAsync(() -> shop.getPrice(product), executor))
//对上一个future的结果额外加工
.map(future -> future.thenApply(Quote::parse))
//thenCompose方法,用第一个future当入参,得下一个future
.map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor)));
}
//使用
public List<String> findPricesFuture(String product) {
//调用上面那个
List<CompletableFuture<String>> priceFutures = findPricesStream(product)
.collect(Collectors.<CompletableFuture<String>>toList());
//真的取数据
return priceFutures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList());
}
thenAccept
改进,不等所有价格都查到了,返回一个就显示一个
public void printPricesStream(String product) {
//开始时间
long start = System.nanoTime();
Stream<CompletableFuture<String>> pricesStream = findPricesStream(product);
CompletableFuture[] futures = pricesStream
//thenAccept 来一个就处理一个,打印显示出来 不等了
.map(f -> f.thenAccept(s -> System.out.println(s + " (done in " + ((System.nanoTime() - start) / 1_000_000) + " msecs)")))
.toArray(size -> new CompletableFuture[size]);
//等全部结束,有个返回就行的话 改成anyOf
CompletableFuture.allOf(futures).join();
//打印:提醒一下全部结束了
System.out.println("All shops have now responded in " + ((System.nanoTime() - start) / 1_000_000) + " msecs");
}
thenCombine
要把查到的欧元换成美元,需要去查汇率,查汇率不需要依赖其他结果,一边拿到欧元价格,一边查到欧元美元汇率,得到最终美元价格,需要合并2个独立的CompletableFuture,得到一个CompletableFuture
CompletableFuture<Double> futurePriceInUSD =
//第一个异步
CompletableFuture.supplyAsync(() -> shop.getPrice(product))
.thenCombine(
//参数1 是一个CompletableFuture
CompletableFuture.supplyAsync( () -> ExchangeService.getRate(Money.EUR, Money.USD)),
//参数2 是怎么融合2个异步结果
(price, rate) -> price * rate
);
