了解CompletableFuture

---

為什么要引入CompletableFuture

為了獲得最佳性能表現(xiàn)，可能就需要仔細(xì)規(guī)劃業(yè)務(wù)流程中的各個(gè)步驟的編排。而Java8就提供了這樣一個(gè)即用容器來(lái)連接一系列任務(wù)。

CompletableFuture 如何來(lái)構(gòu)建一個(gè)任務(wù)鏈呢？這里就不先介紹API了，之前看應(yīng)用場(chǎng)景。

目前有一個(gè)需求：

? 我們需要從 數(shù)據(jù)中臺(tái)中獲取 安徽、江蘇、浙江三省內(nèi)所有企業(yè)信息 及其 子公司 的信息，并按 **子公司數(shù)量 **進(jìn)行排序。

我從觀察到的代碼（finchinaApp）中來(lái)看，如果遇到這種類(lèi)似的需求，基本會(huì)采用多線程池 并發(fā)的查詢(xún)安徽、江蘇、浙江三省的公司，然后線程池內(nèi)線程都結(jié)束執(zhí)行，再對(duì)所有的結(jié)果進(jìn)行排序。

當(dāng)然在這種場(chǎng)景下，使用FutureTask 或者 線程池 +

CountDownLatch countDownLatch = new CountDownLatch(3); new Thread(() - >{//查詢(xún)安徽countDownLatch.countDown(); }).start(); new Thread(() - >{//查詢(xún)江蘇countDownLatch.countDown(); }).start(); new Thread(() - >{//查詢(xún)浙江countDownLatch.countDown(); }).start(); try {countDownLatch.await(); } catch (InterruptedException e) {//日志記錄 } MultiTaskScheduler multiTaskScheduler = new MultiTaskScheduler(); multiTaskScheduler.add(()->{//查詢(xún)安徽 }) multiTaskScheduler.add(()->{//查詢(xún)江蘇 }) multiTaskScheduler.add(()->{//查詢(xún)浙江 }) try {multiTaskScheduler.start(true, 10000); } catch (InterruptedException e) {// + 日志Thread.currentThread().interrupt(); } finally {multiTaskScheduler.shutdown(); }

可以觀察到，該使用場(chǎng)景下，不需要引入CompletableFuture。如果是以下場(chǎng)景呢？

使用多線程 從 redis中獲取 全國(guó)各個(gè)省份內(nèi)所有公司的信息 ，如果reids中缺少某個(gè)省份的信息，則選擇從數(shù)據(jù)庫(kù)中獲取，并將獲取的數(shù)據(jù)異步更新到redis中。在以往我們就需要通過(guò)調(diào)用線程池submit方法將Future返回的結(jié)果 存放在一個(gè)List中，然后通過(guò)for循環(huán)get()得到結(jié)果，然后通過(guò)結(jié)果的不同再進(jìn)行不同的處理。

校驗(yàn)省內(nèi)中所有公司的稅務(wù)報(bào)表，如果某個(gè)公司校驗(yàn)過(guò)程中出現(xiàn)異常，則記錄日志，并且中止整個(gè)省份所有公司的稅務(wù)校驗(yàn)。

ExecutorService threadPool = Executors.newFixedThreadPool(10); List<Future<Boolean>> result = new ArrayList<>(); result.add(threadPool.submit(() -> {System.out.println("執(zhí)行f1");try {Thread.sleep(10000);} catch (InterruptedException e) {e.printStackTrace();}return true; })); result.add(threadPool.submit(() -> {System.out.println("執(zhí)行f2");try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}return false; })); for(int i=0;i<result.size();i++){System.out.println(result.get(i).get()); // main線程會(huì)在此阻塞，先獲取f1的結(jié)果，再獲取f2的結(jié)果 } //實(shí)際上可能f2的結(jié)果早于f1得到，但是主線程仍會(huì)阻塞著等待f1。各個(gè)線程的回調(diào)處理，往往會(huì)取決于運(yùn)行最久的線程。時(shí)間浪費(fèi)嚴(yán)重。

簡(jiǎn)單說(shuō)明：

1. 如果業(yè)務(wù)中每個(gè)線程需要線程結(jié)束后立即進(jìn)行處理，而非等待其他線程都結(jié)束再進(jìn)行操作，則需要使用CompletableFuture。 2. 如果 業(yè)務(wù)1 和 業(yè)務(wù)2 并行執(zhí)行，任意一個(gè)任務(wù)失敗/成功，則標(biāo)志整個(gè)業(yè)務(wù)失敗/成功，則需要使用CompletableFuture 3. 如果 每個(gè)線程中有較為復(fù)雜的異步任務(wù)編排(組合多個(gè)Future并行執(zhí)行結(jié)果)，則建議使用CompletableFuture，因?yàn)榭梢院?jiǎn)化代碼，同時(shí)極其優(yōu)雅。

CompletableFuture的優(yōu)點(diǎn)就在于 :

? ①有回調(diào)函數(shù)，可以在執(zhí)行完成后自動(dòng)調(diào)用回調(diào)處理邏輯；

? ②其提供的四十多個(gè)API可以?xún)?yōu)雅的編排任務(wù)鏈（提供異步任務(wù)完成后的鏈?zhǔn)秸{(diào)用）；

? ③具備異常處理機(jī)制

CompletableFuture的使用

使用誤區(qū)1：CompletableFuture所有的方法都有額外以Async結(jié)尾的方法。此類(lèi)方法的作用在于不是說(shuō) 后續(xù)任務(wù)可以在前序任務(wù)還沒(méi)執(zhí)行完就可以運(yùn)行。舉例：

thenRun(Runnable action) 當(dāng)上一個(gè)任務(wù)結(jié)束后，此任務(wù)沿用上一個(gè)任務(wù)的線程池。

thenRunAsync(Runnable action) 當(dāng)上一個(gè)任務(wù)結(jié)束后，此任務(wù)使用默認(rèn)的ForkJoinPool線程池

thenRunAsync(Runnable action，Executor executor)，當(dāng)上一個(gè)任務(wù)結(jié)束后，此任務(wù)使用自定義線程池（推薦）

/**建議用自定義線程池。CompletableFuture在未指定線程池的情況下，默認(rèn)使用 ForkJoinPool而ForkJoinPool中線程是守護(hù)線程 */ExecutorService threadPool = Executors.newFixedThreadPool(10);CompletableFuture cf1 = new CompletableFuture().supplyAsync(() -> {//執(zhí)行業(yè)務(wù)1return c; },threadPool).thenApplyAsync(c -> {//執(zhí)行業(yè)務(wù)2return c; },threadPool).thenAcceptAsync(c ->{c.stream().sorted(Comparator.comparing(Company::getSubsidiaries)); });

總結(jié)

以上是生活随笔為你收集整理的了解CompletableFuture的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。