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"
示例:
计算商店的商品的售价，其中：售价=价格*利率，价格、利率需要分别从其它系统获取。
商品的售价.png
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)
网友评论