CompletableFuture：組合式異步編程

讓代碼免受阻塞之苦

使用定制的執(zhí)行器

就這個主題而言，明智的選擇似乎是創(chuàng)建一個配有線程池的執(zhí)行器，線程池中線程的數(shù)目取決于你預(yù)計你的應(yīng)用需要處理的負(fù)荷，但是該如何選擇合適的線程數(shù)目呢？

調(diào)整線程池的大小
《Java并發(fā)編程實戰(zhàn)》一書中，Brian Goetz和合著者們?yōu)榫€程池大小
的優(yōu)化提供了不少中肯的建議。這非常重要，如果線程池中線程的數(shù)量過多，最終它們會競爭
稀缺的處理器和內(nèi)存資源，浪費大量的時間在上下文切換上。反之，如果線程的數(shù)目過少，正
如你的應(yīng)用所面臨的情況，處理器的一些核可能就無法充分利用。Brian Goetz建議，線程池大
小與處理器的利用率之比可以使用下面的公式進(jìn)行估算：
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ù)器超負(fù)荷而崩潰，你還是需要設(shè)置一個上限，比如100個線程。代碼清單如下所示（為“最優(yōu)價格查詢器”應(yīng)用定制的執(zhí)行器）：

private final Executor executor = Executors.newFixedThreadPool(Math.min(shops.size(), 100), //創(chuàng)建一個線程池，線程池中線程的數(shù)目為100和商店數(shù)目二者中較小的一個值new ThreadFactory() { public Thread newThread(Runnable r) { Thread t = new Thread(r); t.setDaemon(true); //使用守護(hù)線程——這種方式不會阻止程序的關(guān)停return t; }
}); 

注意，你現(xiàn)在正創(chuàng)建的是一個由守護(hù)線程構(gòu)成的線程池。Java程序無法終止或者退出一個正在運行中的線程，所以最后剩下的那個線程會由于一直等待無法發(fā)生的事件而引發(fā)問題。與此相反，如果將線程標(biāo)記為守護(hù)進(jìn)程，意味著程序退出時它也會被回收。這二者之間沒有性能上的差異。現(xiàn)在，你可以將執(zhí)行器作為第二個參數(shù)傳遞給supplyAsync工廠方法了。比如，你現(xiàn)在可以按照下面的方式創(chuàng)建一個可查詢指定商品價格的CompletableFuture對象：

CompletableFuture.supplyAsync(() -> shop.getName() + " price is " + shop.getPrice(product), executor); 

改進(jìn)之后，使用CompletableFuture方案的程序處理5個商店僅耗時1021秒，處理9個商店時耗時1022秒。一般而言，這種狀態(tài)會一直持續(xù)，直到商店的數(shù)目達(dá)到我們之前計算的閾值400。這個例子證明了要創(chuàng)建更適合你的應(yīng)用特性的執(zhí)行器，利用CompletableFutures向其提交任務(wù)執(zhí)行是個不錯的主意。處理需大量使用異步操作的情況時，這幾乎是最有效的策略。

并行——使用流還是CompletableFutures？目前為止，你已經(jīng)知道對集合進(jìn)行并行計算有兩種方式：要么將其轉(zhuǎn)化為并行流，利用map這樣的操作開展工作，要么枚舉出集合中的每一個元素，創(chuàng)建新的線程，在CompletableFuture內(nèi)對其進(jìn)行操作。后者提供了更多的靈活性，你可以調(diào)整線程池的大小，而這能幫助你確保整體的計算不會因為線程都在等待I/O而發(fā)生阻塞。我們對使用這些API的建議如下。
?如果你進(jìn)行的是計算密集型的操作，并且沒有I/O，那么推薦使用Stream接口，因為實
現(xiàn)簡單，同時效率也可能是最高的（如果所有的線程都是計算密集型的，那就沒有必要
創(chuàng)建比處理器核數(shù)更多的線程）。
?反之，如果你并行的工作單元還涉及等待I/O的操作（包括網(wǎng)絡(luò)連接等待），那么使用
CompletableFuture靈活性更好，你可以像前文討論的那樣，依據(jù)等待/計算，或者
W/C的比率設(shè)定需要使用的線程數(shù)。這種情況不使用并行流的另一個原因是，處理流的
流水線中如果發(fā)生I/O等待，流的延遲特性會讓我們很難判斷到底什么時候觸發(fā)了等待。

現(xiàn)在已經(jīng)了解了如何利用CompletableFuture為你的用戶提供異步API，以及如何將一個同步又緩慢的服務(wù)轉(zhuǎn)換為異步的服務(wù)。不過到目前為止，每個Future中進(jìn)行的都是單次的操作。

對多個異步任務(wù)進(jìn)行流水線操作

讓我們假設(shè)所有的商店都同意使用一個集中式的折扣服務(wù)。該折扣服務(wù)提供了五個不同的折扣代碼，每個折扣代碼對應(yīng)不同的折扣率。你使用一個枚舉型變量Discount.Code來實現(xiàn)這一想法，具體代碼如下所示。

public class Discount {public enum Code {NONE(0), SILVER(5), GOLD(10), PLATINUM(15), DIAMOND(20); private final int percentage; Code(int percentage) { this.percentage = percentage; } } // Discount類的具體實現(xiàn)這里暫且不表示
} 

還假設(shè)所有的商店都同意修改getPrice方法的返回格式。getPrice現(xiàn)在以ShopName:price:DiscountCode的格式返回一個String類型的值。示例實現(xiàn)中會返回一個隨機生成的Discount.Code，以及已經(jīng)計算得出的隨機價格：

public String getPrice(String product) {double price = calculatePrice(product); Discount.Code code = Discount.Code.values()[ random.nextInt(Discount.Code.values().length)]; return String.format("%s:%.2f:%s", name, price, code); 
} 
private double calculatePrice(String product) { delay(); return random.nextDouble() * product.charAt(0) + product.charAt(1); 
} 

調(diào)用getPrice方法可能會返回像下面這樣一個String值：

BestPrice:123.26:GOLD