並發與並行
Java 5並發庫主要關注於異步任務的處理，它采用了這樣一種模式，producer線程創建任務並且利用阻塞隊列將其傳遞給任務的consumer。這種模型在Java 7和8中進一步發展，並且開始支持另外一種風格的任務執行，那就是將任務的數據集分解為子集，每個子集都可以由獨立且同質的子任務來負責處理。
這種風格的基礎庫也就是fork/join框架，它允許程序員規定數據集該如何進行分割，並且支持將子任務提交到默認的標准線程池中，也就是“通用的”ForkJoinPool。（在本文中，非全限定的類和接口名指的都是java.util.concurrent包中的類型。）在Java 8中，fork/join並行功能借助並行流的機制變得更加具有可用性。但是，不是所有的問題都適合這種風格的並行處理：所處理的元素必須是獨立的，數據集要足夠大，並且在並行加速方面，每個元素的處理成本要足夠高，這樣才能補償建立fork/join框架所消耗的成本。
同時，Java 8在並行流方面的革新得到了廣泛的關注，這導致大家忽略了並發庫中一項新增的重要功能，那就是CompletableFuture<T>類。本文將會探討CompletableFuture類，有一些系統會依賴於不同類型的異步執行任務，本文將會闡述該類為什么會對這種類型的系統如此重要，並介紹了它是如何補充fork/join風格的並行機制和並行流的。
頁面渲染器
我們的起始點將是“Java Concurrency in Practice”（JCiP）一書中的樣例，這個樣例非常經典地闡述了Java 5中的並發工具類。在JCiP的第6.3節中，Brian Goetz探討了如何開發一個Web頁面的渲染器，對於每個頁面來說，它的任務就是渲染文本，並下載和渲染圖片。圖片的下載會耗費較長的時間，在這段時間內CPU無事可做，只能等待。所以，在渲染頁面時，一個很明顯的策略就是首先初始化所有圖片的下載，然后利用它們完成之前的這段時間渲染頁面文本，最后渲染下載的圖片。
相關贊助商
在JCiP中，第一版本的頁面渲染器使用了Future的理念，這個接口暴露了一些方法，允許客戶端監控任務的執行進度，這里的任務是在一個不同的進程中執行的。在程序清單1中，Callable代表了下載頁面中所有圖片的任務，它被提交到了Executor中，然后返回一個Future對象，通過它就能詢問下載任務的狀態。當主線程渲染完頁面的文本后，會調用Future.get方法，這個方法會一直阻塞直到所有下載的結果均可用為止，在本例中這個結果是以List<ImageData>的形式來表示的。這種方式一個很明顯的缺點在於下載任務的粗粒度性，在所有的圖片下載完成之前，我們一張圖片也不能渲染。接下來，我們看一下如何緩解這個問題。
public void renderPage(CharSequence source) {
List<ImageInfo> info = scanForImageInfo(source);
//創建Callable，它代表了下載所有的圖片
final Callable<List<ImageData>> task = () ->
info.stream()
.map(ImageInfo::downloadImage)
.collect(Collectors.toList());
// 將下載任務提交到executor
Future<List<ImageData>> images = executor.submit(task);
// renderText(source);
try {
// 獲得所有下載的圖片（在所有圖片可用之前會一直阻塞）
final List<ImageData> imageDatas = images.get();
// 渲染圖片
imageDatas.forEach(this::renderImage);
} catch (InterruptedException e) {
// 重新維護線程的中斷狀態
Thread.currentThread().interrupt();
// 我們不需要結果，所以取消任務
images.cancel(true);
} catch (ExecutionException e) {
throw launderThrowable(e.getCause()); }
}程序清單1：使用Future等待所有的圖片下載完成
為了讓這個樣例及其后面的變種易於管理，這里有一個前提條件：我們假設類型ImageInfo（簡單來講，就是一個URL）和ImageData（圖片的二進制數據）以及方法scanForImageInfo、downloadImage、renderText、renderImage、launderThrowable和ImageInfo.downloadImage都已經存在了。實例變量executor是通過ExecutorService類型聲明的並進行了恰當的初始化。在本文中，我將JCiP中最初的樣例利用Java 8 lambda表達式和流進行了現代化。
在這段代碼中，必須要等待所有下載都完成的原因在於，它使用Future接口來代表下載任務，作為異步執行的任務模型，它有很大的局限性。Future允許客戶端詢問任務執行的結果，如果必要的話，將會產生阻塞，另外還可以詢問任務的狀態，判斷它已經完成還是被取消了。但是，Future本身並不能提供回調方法，假設能夠這樣做的話，當每個圖片下載完成的時候，就能通知頁面的渲染線程了。
程序清單2改善了之前樣例中粒度較粗的問題，它將頁面下載的任務提交到了CompletionService類中，這個類的poll和take方法會產生對應的Future實例，這些實例是按照任務完成的順序排列的，而不是像程序清單1那樣任務是按照提交的順序處理的。在ExecutorCompletionService接口的平台實現中，為了實現該功能，每項任務都會包裝在一個FutureTask中，FutureTask是Future的一個實現，它允許提供完成時的回調。Future的回調行為是在ExecutorCompletionService中創建的，完成的任務會封裝到一個隊列中，供客戶端詢問時使用。
public void renderPage(CharSequence source) {
List<ImageInfo> info = scanForImageInfo(source);
CompletionService<ImageData> completionService =
new ExecutorCompletionService<>(executor);
// 將每個下載任務提交到completion service中
info.forEach(imageInfo ->
completionService.submit(imageInfo::downloadImage));
renderText(source);
// 當每個RunnableFuture可用時（並且我們也准備處理它的時候），
// 將它們檢索出來
for (int t = 0; t < info.size(); t++) {
Future<ImageData> imageFuture = completionService.take();
renderImage(imageFuture.get());
}
}
程序清單2：借助CompletionService，當圖片可用時立即將其渲染出來（為了保持簡潔性，省略掉了中斷和錯誤處理的代碼）
CompletableFuture簡介
程序清單2代表了Java 5所能達到的水准，不過2014年之后，在Java中，編寫異步系統的表現性得到了巨大的提升，這是通過引入CompletableFuture (CF)類實現的。這個類是Future的實現，它能夠將回調放到與任務不同的線程中執行，也能將回調作為繼續執行的同步函數，在與任務相同的線程中執行。它避免了傳統回調最大的問題，那就是能夠將控制流分離到不同的事件處理器中，而這是通過允許CF實例與回調方法進行組合形成新的CF來實現的。
作為樣例，可以參考thenAccept方法，它接受一個 Consumer（用戶提供的且沒有返回值的函數）並返回一個新的CF。這個新CF所能達到的效果就是在最初CF完成時所得到的結果上，運用Consumer。與很多其他的CF方法類似，thenAccept有兩個變種形式，在第一個中，Consumer會由通用fork/join池中的某一個線程來執行；在第二個中，它會由Executor中的某一個線程來負責執行，而Executor是我們在調用時提供的。這形成了三種重載形式：同步運行、在ForkJoinPool中異步運行以及在調用時所提供的線程池中異步運行，CompletableFuture中有近60個方法，上述的這三種重載形式占了絕大多數。
如下是thenAccept的一個樣例，借助它重新實現了頁面渲染器的功能：
public void renderPage(CharSequence source) {
List<ImageInfo> info = scanForImageInfo(source);
info.forEach(imageInfo ->
CompletableFuture
.supplyAsync(imageInfo::downloadImage)
.thenAccept(this::renderImage));
renderText(source);
}
程序清單3：使用CompletableFuture來實現頁面渲染功能
盡管程序清單3比前面的形式更加簡潔，但是我們需要練習一下才能更好地閱讀它。工廠方法supplyAsync返回一個新的CF，它會在通用的 ForkJoinPool中運行指定的Supplier，完成時，Supplier的結果將會作為CF的結果。方法thenAccept會返回一個新的CF，它將會執行指定的Consumer，在本例中也就是渲染給定的圖片，即supplyAsync方法所產生的CF的結果。
需要澄清的是，thenAccept並不是將CF與函數組合起來的唯一方式。將CF與函數組合起來可以接受如下的參數：
- 應用於
CF操作結果的函數。此時，可以采用的方法包括：thenCompose：針對返回值為CompletableFuture的函數；thenApply：針對返回值為其他類型的函數；thenAccept：針對返回值為void的函數；
Runnable。通過thenRun方法，可以接受Runnable參數；- 函數在處理的過程中，可能正常結束也可能異常退出。
CF能夠通過方法來分別組合這兩種情況：handle，針對接受一個值和一個Throwable，並有返回值的函數；whenComplete，針對接受一個值和一個Throwable，並返回void的函數。
擴展頁面渲染器
擴展該樣例能夠闡述CompletableFuture的其他特性。比如，當圖片下載超時或失敗時，我們想使用一個圖標作為可見的指示器。CF暴露了一個名為get(long, TimeUnit)的方法，如果 CF在指定的時間內沒有完成的話，將會拋出TimeoutException異常。我們可以使用它來定義一個函數，這個函數會將 ImageInfo轉換為ImageData （程序清單4）。
Function<ImageInfo, ImageData> infoToData = imageInfo -> {
CompletableFuture<ImageData> imageDataFuture =
CompletableFuture.supplyAsync(imageInfo::downloadImage, executor);
try {
return imageDataFuture.get(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
imageDataFuture.cancel(true);
return ImageData.createIcon(e);
} catch (ExecutionException e) {
throw launderThrowable(e.getCause());
} catch (TimeoutException e) {
return ImageData.createIcon(e);
}
}
程序清單4：使用CompletableFuture.get來實現超時
現在，頁面可以通過連續調用infoToData來進行渲染。其中每個調用都會同步返回一個下載的圖片，所以要並行下載的話，需要為它們各自創建一個新的異步任務。要實現這一功能，合適的工廠方法是CompletableFuture.runAsync()，它與supplyAsync類似，但是接受的參數是Runnable而不是Supplier：
public void renderPage(CharSequence source) throws InterruptedException {
List<ImageInfo> info = scanForImageInfo(source);
info.forEach(imageInfo ->
CompletableFuture.runAsync(() ->
renderImage(infoToData.apply(imageInfo)), executor));
}
現在，我們考慮進一步的需求，當所有的請求完成或超時后，在頁面上顯示一個指示器，如果對應的所有CompletableFuture都從join方法中返回，就能表示出現了這種場景。靜態方法allOf就是為這種需求而提供的，它能夠創建一個返回值為空的CompletableFuture，當其所有的組件均完成時，它也會達到完成狀態。（join方法通常用來返回某個CF的結果，為了查看allOf方法所組合起來的所有CF的結果，必須要對其進行單獨地查詢。）
public void renderPage(CharSequence source) {
List<ImageInfo> info = scanForImageInfo(source);
CompletableFuture[] cfs = info.stream()
.map(ii -> CompletableFuture.runAsync(
() -> renderImage(mapper.apply(ii)), executor))
.toArray(CompletableFuture[]::new);
CompletableFuture.allOf(cfs).join();
renderImage(ImageData.createDoneIcon());
}
聯合多個CompletableFuture
另外一組方法允許將多個CF聯合在一起。我們已經看見過靜態方法allOf，當其所有的組件均完成時，它就會處於完成狀態，與之對應的方法也就是anyOf，返回值同樣是void，當其任意一個組件完成時，它就會完成。除了這兩個方法以外，這個組中其他的方法都是實例方法，它們能夠將receiver按照某種方式與另外一個CF聯合在一起，然后將結果傳遞到給定的函數中。
為了展現它們是如何運行的，我們擴展一下JCiP中的另一個例子，這是一個旅行預訂的門戶，我們將互相關聯的訂購過程記錄在TripPlan對象中，它包含了總價以及所使用服務供應商的列表：
interface TripPlan {
List<ServiceSupplier> getSuppliers();
int getPrice();
TripPlan combine(TripPlan);
}
ServiceSupplier（比如說某個航線或酒店）能夠創建一個TripPlan：（當然，在現實中，ServiceSupplier.createPlan 將會接受參數，來反映對應的目的地、旅行等級等信息。）
interface ServiceSupplier {
TripPlan createPlan();
String getAlliance(); // 稍后使用
}
為了選擇最佳的旅行計划，需要查詢每個服務供應商為我們的旅行所給定的規划，然后使用Comparator來對比每個規划結果，這個Comparator反映了我們的選擇標准（在本例中，只是簡單的選擇價格最低者）：
TripPlan selectBestTripPlan(List<ServiceSupplier> serviceList) {
List<CompletableFuture<TripPlan>> tripPlanFutures = serviceList.stream()
.map(svc -> CompletableFuture.supplyAsync(svc::createPlan, executor))
.collect(toList());
return tripPlanFutures.stream()
.min(Comparator.comparing(cf -> cf.join().getPrice()))
.get().join();
}
請注意中間的collect操作，在流處理里面，由於中間操作的延遲性（laziness of intermediate operation），它就變得非常必要了。如果沒有它的話，流的終止操作（terminal operation）將會是min，它如果要執行的話，首先需要針對tripPlanFutures的每個元素執行join操作。如上述的代碼所示，我們並沒有這樣做，終止操作是collect，它會將map操作所形成的CF值累積起來，這個過程中沒有阻塞，因此允許底層的任務並發執行。
如果獲取航線和酒店最佳旅行計划的任務是獨立的，那么我們會希望它們能夠同時初始化，就像前文所述的圖片下載一樣。要將兩個CF按照這種方式聯合在一起，我們需要使用CompletableFuture.thenCombine方法，它會並行地執行receiver以及所提供的CF，然后將它們的結果使用給定的函數組合起來（在這里，假設變量 airlines、hotels和（稍后使用的）cars都是以List<TravelService> 類型進行聲明的，並且已經進行了恰當的初始化）：
CompletableFuture
.supplyAsync(() -> selectBestTripPlan(airlines))
.thenCombine(
CompletableFuture.supplyAsync(() -> selectBestTripPlan(hotels)),
TripPlan::combine);
對這個樣例進行擴展，我們將會學到更多的內容。假設每個服務供應商都屬於某一個旅行聯盟（travel alliance），通過String類型的屬性alliance來表示。在獨立訂購完航線和酒店后，我們將會確定它們是否屬於同一個聯盟，如果是的話，那么只有屬於同一聯盟的租車服務，才在我們的考慮范圍之內：
private TripPlan addCarHire(TripPlan p) {
List<String> alliances = p.getSuppliers().stream()
.map(ServiceSupplier::getAlliance)
.distinct()
.collect(toList());
if (alliances.size() == 1) {
return p.combine(selectBestTripPlan(cars, alliances.get(0)));
} else {
return p.combine(selectBestTripPlan(cars));
}
}
selectBestTripPlan方法新的重載形式將會接受一個String類型作為偏愛的聯盟，如果這個值存在的話，會使用它來過濾流中的服務：
private TripPlan selectBestTripPlan(
List<ServiceSupplier> serviceList, String favoredAlliance) {
List<CompletableFuture<TripPlan>> tripPlanFutures = serviceList.stream()
.filter(ts ->
favoredAlliance == null || ts.getAlliance().equals(favoredAlliance))
.map(svc -> CompletableFuture.supplyAsync(svc::createPlan, executor))
.collect(toList());
...
}
在本例中，選擇租車服務的CF要依賴於航班和酒店預訂任務組合所形成的CF。只有航班和酒店都預訂之后，它才能完成。實現這種關聯關系的方法就是thenCompose：
CompletableFuture.supplyAsync(() -> selectBestTripPlan(airlines))
.thenCombine(
CompletableFuture.supplyAsync(() -> selectBestTripPlan(hotels)),
TripPlan::combine)
.thenCompose(p -> CompletableFuture.supplyAsync(() -> addCarHire(p)));
預訂航班和酒店聯合形成的CF會執行，並且它的結果，也就是聯合后的TripPlan，將會作為thenCompose函數參數的輸入。結果形成的CF非常簡潔地封裝了不同異步服務之間的依賴關系。這段代碼如此簡潔的原因在於，盡管thenCompose聯合了兩個CF，但是它所返回的並不是我們預期的CompletableFuture<CompletableFuture<TripPlan>>，而是CompletableFuture<TripPlan>。所以，不管在創建CF的時候使用了多少層級的組合，它並不是嵌套的，而是扁平的，要獲取它的結果只需要一步操作。這是monad“綁定（bind）”操作（這個名稱來源於Haskell）的特性，CF就是這種monad，並且闡明了monad一些非常積極的特征：比如，在本例中，我們能夠按照函數式的形式進行編寫，如果沒有這項功能的話，就需要在各個回調中非常繁瑣地顯式編寫任務定義。
thenCombine方法只是將兩個CF聯合起來的方法之一，其他的方法包括：
thenAcceptBoth：與thenCombine類似，但是它接受一個返回值為void的函數；runAfterBoth：接受一個Runnable，在兩個CF都完成后執行；applyToEither：接受一個一元函數（unary function），會將首先完成的CF的結果提供給它；acceptEither：與applyToEither類似，接受一個一元函數，但是結果為void；runAfterEither：接受一個Runnable，在其中一個CF完成后就執行。
結論
我們不可能在一篇短文中，完整地闡述像CompletableFuture這樣的API，但是我希望這里的樣例能夠讓你對它所能實現的並發編程形式有一個直觀印象。將CompletableFuture與其他CompletableFuture組合，以及與其他函數組合，能夠為多項任務構建類似流水線的方案，這樣能夠控制同步和異步執行以及它們之間的依賴。你想更加詳細了解的內容可能會包括異常處理、選擇和配置executor的實際經驗以及設計異步API所面臨的挑戰。
我希望已經解釋清楚了Java 8所提供的兩種異步編程風格之間的聯系。在使用fork/join並行機制（包括並行流）的場景中，能夠非常高效的將工作內容進行跨核心分發。但是，它的適用條件卻非常有限：數據集很大並且能夠高效地分割，對某個數據元素的操作與其他元素是（相對）獨立的，這些操作的成本應該是比較高昂的，並且應該是CPU密集型的。如果這些條件無法滿足的話，尤其是如果你的任務會花費很多時間阻塞在I/O或網絡請求上的話，那么 CompletableFuture是更好的替代方案。作為Java程序員，我們非常幸運地有這樣一個平台庫，它將這些補充的方式集成在了一起。