Future 接口
Future接口是Java5引入的萧落，提供了異步執(zhí)行任務(wù)的能力
ExecutorService executor = Executors.newCachedThreadPool();
Future<Integer> future = executor.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
//異步執(zhí)行耗時的操作
return dongSomeLongComputation();
}
});
doSomeThingElse();//異步操作的同時執(zhí)行其它操作
try {
future.get(1, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
對于結(jié)果的獲取卻是很不方便蹲盘，只能通過阻塞或者輪詢的方式得到任務(wù)的結(jié)果喊熟。阻塞的方式顯然和我們的異步編程的初衷相違背，輪詢的方式又會耗費無謂的CPU資源达皿，而且也不能及時地得到計算結(jié)果天吓。
例如：
1、將兩個異步計算合并成一個峦椰，其中一個依賴另一個的結(jié)果
2龄寞、等待Future集合中的任務(wù)完成，或等待Future集合中執(zhí)行最快的任務(wù)完成
CompletableFuture
在Java 8中, 新增加了一個包含50個方法左右的類: CompletableFuture汤功，提供了非常強大的Future的擴展功能物邑，可以幫助我們簡化異步編程的復(fù)雜性，提供了函數(shù)式編程的能力冤竹，可以通過回調(diào)的方式處理計算結(jié)果拂封，并且提供了轉(zhuǎn)換和組合CompletableFuture的方法。
創(chuàng)建CompletableFuture對象鹦蠕。
基本用法
創(chuàng)建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(() -> {
//長時間的計算任務(wù)
return "·00";
});
future.get();
轉(zhuǎn)換
CompletableFuture可以作為monad(單子)和functor。由于回調(diào)風(fēng)格的實現(xiàn)钟病，我們不必因為等待一個計算完成而阻塞著調(diào)用線程萧恕，而是告訴CompletableFuture當計算完成的時候請執(zhí)行某個function刚梭。而且我們還可以將這些操作串聯(lián)起來，或者將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"
示例:
計算商店的商品的售價朴读，其中：售價=價格*利率，價格走趋、利率需要分別從其它系統(tǒng)獲取衅金。
java7實現(xiàn);
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實現(xiàn)：
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);
使用定制的執(zhí)行器
執(zhí)行器的默認線程數(shù)量為：Runtime.getRuntime().availableProcessors()
調(diào)整線程池的大小
《Java并發(fā)編程實戰(zhàn)》一書中，BrianGoetz和合著者們?yōu)榫€程池大小的優(yōu)化提供了不少中肯的建議簿煌。這非常重要氮唯，如果線程池中線程的數(shù)量過多，最終它們會競爭稀缺的處理器和內(nèi)存資源姨伟，浪費大量的時間在上下文切換上惩琉。反之，如果線程的數(shù)目過少夺荒，正如你的應(yīng)用所面臨的情況瞒渠，處理器的一些核可能就無法充分利用。BrianGoetz建議技扼，線程池大小與處理器的利用率之比可以使用下面的公式進行估算：
Nthreads = Ncpu * Ucpu * (1 + W/C)
Ncpu是處理器的核的數(shù)目伍玖，可以通過Runtime.getRuntime().availableProcessors()得到
Ucpu是期望的CPU利用率（該值應(yīng)該介于0和1之間）
W/C是等待時間與計算時間的比率
你的應(yīng)用99%的時間都在等待商店的響應(yīng)，所以估算出的W/C比率為100淮摔。這意味著如果你期望的CPU利用率是100%私沮，你需要創(chuàng)建一個擁有400個線程的線程池。實際操作中和橙，如果你創(chuàng)建的線程數(shù)比商店的數(shù)目更多仔燕，反而是一種浪費，因為這樣做之后魔招，你線程池中的有些線程根本沒有機會被使用晰搀。出于這種考慮，我們建議你將執(zhí)行器使用的線程數(shù)办斑，與你需要查詢的商店數(shù)目設(shè)定為同一個值外恕，這樣每個商店都應(yīng)該對應(yīng)一個服務(wù)線程。不過乡翅，為了避免發(fā)生由于商店的數(shù)目過多導(dǎo)致服務(wù)器超負荷而崩潰鳞疲，你還是需要設(shè)置一個上限，比如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);//使用守護線程——這種方式不會阻止程序的關(guān)停
return t;
}
});
//單筆獲取結(jié)果
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
//長時間的計算任務(wù)
return "1·00";
}, executor);
future.get();
//批量獲取結(jié)果
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)
