Future 接口
Future接口是Java5引入的,提供了异步执行任务的能力
ExecutorService executor = Executors.newCachedThreadPool();
Future<Integer> future = executor.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
//异步执行耗时的操作
return dongSomeLongComputation();
}
});
doSomeThingElse();//异步操作的同时执行其它操作
try {
future.get(1, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
对于结果的获取却是很不方便,只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背,轮询的方式又会耗费无谓的CPU资源,而且也不能及时地得到计算结果。
例如:
1、将两个异步计算合并成一个,其中一个依赖另一个的结果
2、等待Future集合中的任务完成,或等待Future集合中执行最快的任务完成
CompletableFuture
在Java 8中, 新增加了一个包含50个方法左右的类: CompletableFuture,提供了非常强大的Future的扩展功能,可以帮助我们简化异步编程的复杂性,提供了函数式编程的能力,可以通过回调的方式处理计算结果,并且提供了转换和组合CompletableFuture的方法。
创建CompletableFuture对象。
基本用法
创建CompletableFuture对象。
public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
public T get()
public T get(long timeout, TimeUnit unit)
public T getNow(T valueIfAbsent)
public T join()
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
//长时间的计算任务
return "·00";
});
future.get();
转换
CompletableFuture可以作为monad(单子)和functor。由于回调风格的实现,我们不必因为等待一个计算完成而阻塞着调用线程,而是告诉CompletableFuture当计算完成的时候请执行某个function。而且我们还可以将这些操作串联起来,或者将CompletableFuture组合起来。
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
return 100;
});
CompletableFuture<String> f = future.thenApplyAsync(i -> i * 10).thenApply(i -> i.toString());
System.out.println(f.get()); //"1000"
示例:
计算商店的商品的售价,其中:售价=价格*利率,价格、利率需要分别从其它系统获取。
java7实现;
ExecutorService executor = Executors.newCachedThreadPool();
final Future<Double> futureRate = executor.submit(new Callable<Double>() {
public Double call() {
return ExchangeService.getRate(Money.EUR, Money.US);
}
});
Future<Double> futurePriceInUSD = executor.submit(new Callable<Double>() {
public Double call() {
try {
double priceInEUR = shop.getPrice(product);
return priceInEUR * futureRate.get();
} catch (InterruptedException | ExecutionException e) {
throw new RuntimeException(e.getMessage(), e);
}
}
});
java8实现:
CompletableFuture<String> futurePriceInUSD =
CompletableFuture.supplyAsync(() -> shop.getPrice(product))
.thenCombine(
CompletableFuture.supplyAsync(
() -> ExchangeService.getRate(Money.EUR, Money.USD)),
(price, rate) -> price * rate
).thenApply(price -> shop.getName() + " price is " + price);
使用定制的执行器
执行器的默认线程数量为:Runtime.getRuntime().availableProcessors()
调整线程池的大小
《Java并发编程实战》一书中,BrianGoetz和合著者们为线程池大小的优化提供了不少中肯的建议。这非常重要,如果线程池中线程的数量过多,最终它们会竞争稀缺的处理器和内存资源,浪费大量的时间在上下文切换上。反之,如果线程的数目过少,正如你的应用所面临的情况,处理器的一些核可能就无法充分利用。BrianGoetz建议,线程池大小与处理器的利用率之比可以使用下面的公式进行估算:
Nthreads = Ncpu * Ucpu * (1 + W/C)
Ncpu是处理器的核的数目,可以通过Runtime.getRuntime().availableProcessors()得到
Ucpu是期望的CPU利用率(该值应该介于0和1之间)
W/C是等待时间与计算时间的比率
你的应用99%的时间都在等待商店的响应,所以估算出的W/C比率为100。这意味着如果你期望的CPU利用率是100%,你需要创建一个拥有400个线程的线程池。实际操作中,如果你创建的线程数比商店的数目更多,反而是一种浪费,因为这样做之后,你线程池中的有些线程根本没有机会被使用。出于这种考虑,我们建议你将执行器使用的线程数,与你需要查询的商店数目设定为同一个值,这样每个商店都应该对应一个服务线程。不过,为了避免发生由于商店的数目过多导致服务器超负荷而崩溃,你还是需要设置一个上限,比如100个线程。代码清单如下所示。
public final static Executor executor = Executors.newFixedThreadPool(100, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setDaemon(true);//使用守护线程——这种方式不会阻止程序的关停
return t;
}
});
//单笔获取结果
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
//长时间的计算任务
return "1·00";
}, executor);
future.get();
//批量获取结果
List<CompletableFuture<String>> priceFutures = shops.stream()
.map(shop -> CompletableFuture.supplyAsync(() -> shop.getName() + " price is " + shop.getPrice(product), executor))
.collect(Collectors.toList());
List<String> prices = priceFutures.stream()
.map(CompletableFuture::join)