（給ImportNew加星標，提高Java技能）
CompletableFuture是jdk1.8引入的實現類。擴展了Future和CompletionStage，是一個可以在任務完成階段觸發一些操作Future。簡單的來講就是可以實現異步回調。
對於jdk1.5的Future，雖然提供了異步處理任務的能力，但是獲取結果的方式很不優雅，還是需要通過阻塞（或者輪訓）的方式。如何避免阻塞呢？其實就是註冊回調。
業界結合觀察者模式實現異步回調。也就是當任務執行完成後去通知觀察者。比如Netty的ChannelFuture，可以通過註冊監聽實現異步結果的處理。
Netty的ChannelFuturepublic Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) { checkNotNull(listener, "listener"); synchronized (this) { addListener0(listener); } if (isDone()) { notifyListeners(); } return this; } private boolean setValue0(Object objResult) { if (RESULT_UPDATER.compareAndSet(this, null, objResult) || RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) { if (checkNotifyWaiters()) { notifyListeners(); } return true; } return false; }
通過addListener方法註冊監聽。如果任務完成，會調用notifyListeners通知。
CompletableFuture通過擴展Future，引入函數式編程，通過回調的方式去處理結果。
CompletableFuture的功能主要體現在他的CompletionStage。
可以實現如下等功能
消費和運行的區別：
消費使用執行結果。運行則只是運行特定任務。具體其他功能大家可以根據需求自行查看。
CompletableFuture藉助CompletionStage的方法可以實現鏈式調用。並且可以選擇同步或者異步兩種方式。
這裡舉個簡單的例子來體驗一下他的功能。
public static void thenApply() { ExecutorService executorService = Executors.newFixedThreadPool(2); CompletableFuture cf = CompletableFuture.supplyAsync(() -> { try { // Thread.sleep(2000); } catch (Exception e) { e.printStackTrace(); } System.out.println("supplyAsync " + Thread.currentThread().getName()); return "hello"; }, executorService).thenApplyAsync(s -> { System.out.println(s + "world"); return "hhh"; }, executorService); cf.thenRunAsync(() -> { System.out.println("ddddd"); }); cf.thenRun(() -> { System.out.println("ddddsd"); }); cf.thenRun(() -> { System.out.println(Thread.currentThread()); System.out.println("dddaewdd"); }); }執行結果
supplyAsync pool-1-thread-1 helloworld ddddd ddddsd Thread[main,5,main] dddaewdd 根據結果我們可以看到會有序執行對應任務。
注意：
如果是同步執行cf.thenRun。他的執行線程可能main線程，也可能是執行源任務的線程。如果執行源任務的線程在main調用之前執行完了任務。那麼cf.thenRun方法會由main線程調用。
這裡說明一下，如果是同一任務的依賴任務有多個：
如果這些依賴任務都是同步執行。那麼假如這些任務被當前調用線程（main）執行，則是有序執行，假如被執行源任務的線程執行，那麼會是倒序執行。因為內部任務數據結構為LIFO。
如果這些依賴任務都是異步執行，那麼他會通過異步線程池去執行任務。不能保證任務的執行順序。
上面的結論是通過閱讀源代碼得到的。下面我們深入源代碼。
創建的方法有很多，甚至可以直接new一個。我們來看一下supplyAsync異步創建的方法。public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) { return asyncSupplyStage(screenExecutor(executor), supplier); } static Executor screenExecutor(Executor e) { if (!useCommonPool && e == ForkJoinPool.commonPool()) return asyncPool; if (e == null) throw new NullPointerException(); return e; }入參Supplier，帶返回值的函數。如果是異步方法，並且傳遞了執行器，那麼會使用傳入的執行器去執行任務。否則採用公共的ForkJoin並行線程池，如果不支持並行，新建一個線程去執行。
這裡我們需要注意ForkJoin是通過守護線程去執行任務的。所以必須有非守護線程的存在才行。
asyncSupplyStage方法static <U> CompletableFuture<U> asyncSupplyStage(Executor e, Supplier<U> f) { if (f == null) throw new NullPointerException(); CompletableFuture<U> d = new CompletableFuture<U>(); e.execute(new AsyncSupply<U>(d, f)); return d; }
這裡會創建一個用於返回的CompletableFuture。
然後構造一個AsyncSupply，並將創建的CompletableFuture作為構造參數傳入。
那麼，任務的執行完全依賴AsyncSupply。
AsyncSupply#runpublic void run() { CompletableFuture<T> d; Supplier<T> f; if ((d = dep) != null && (f = fn) != null) { dep = null; fn = null; if (d.result == null) { try { d.completeValue(f.get()); } catch (Throwable ex) { d.completeThrowable(ex); } } d.postComplete(); } }該方法會調用Supplier的get方法。並將結果設置到CompletableFuture中。我們應該清楚這些操作都是在異步線程中調用的。
d.postComplete方法就是通知任務執行完成。觸發後續依賴任務的執行，也就是實現CompletionStage的關鍵點。
在看postComplete方法之前我們先來看一下創建依賴任務的邏輯。
thenAcceptAsync方法public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) { return uniAcceptStage(asyncPool, action); } private CompletableFuture<Void> uniAcceptStage(Executor e, Consumer<? super T> f) { if (f == null) throw new NullPointerException(); CompletableFuture<Void> d = new CompletableFuture<Void>(); if (e != null || !d.uniAccept(this, f, null)) { # 1 UniAccept<T> c = new UniAccept<T>(e, d, this, f); push(c); c.tryFire(SYNC); } return d; }
上面提到過。thenAcceptAsync是用來消費CompletableFuture的。該方法調用uniAcceptStage。
uniAcceptStage邏輯：
構造一個CompletableFuture，主要是為了鏈式調用。
如果為異步任務，直接返回。因為源任務結束後會觸發異步線程執行對應邏輯。
如果為同步任務（e==null），會調用d.uniAccept方法。這個方法在這裡邏輯：如果源任務完成，調用f，返回true。否則進入if代碼塊（Mark 1）。
如果是異步任務直接進入if（Mark 1）。
Mark1邏輯：
構造一個UniAccept，將其push入棧。這裡通過CAS實現樂觀鎖實現。
調用c.tryFire方法。
final CompletableFuture<Void> tryFire(int mode) { CompletableFuture<Void> d; CompletableFuture<T> a; if ((d = dep) == null || !d.uniAccept(a = src, fn, mode > 0 ? null : this)) return null; dep = null; src = null; fn = null; return d.postFire(a, mode); }會調用d.uniAccept方法。其實該方法判斷源任務是否完成，如果完成則執行依賴任務，否則返回false。如果依賴任務已經執行，調用d.postFire，主要就是Fire的後續處理。根據不同模式邏輯不同。
這裡簡單說一下，其實mode有同步異步，和迭代。迭代為了避免無限遞歸。
這裡強調一下d.uniAccept方法的第三個參數。
如果是異步調用（mode>0），傳入null。否則傳入this。
區別看下面代碼。c不為null會調用c.claim方法。
try { if (c != null && !c.claim()) return false; @SuppressWarnings("unchecked") S s = (S) r; f.accept(s); completeNull(); } catch (Throwable ex) { completeThrowable(ex); } final boolean claim() { Executor e = executor; if (compareAndSetForkJoinTaskTag((short)0, (short)1)) { if (e == null) return true; executor = null; // disable e.execute(this); } return false; }如果異步線程為null。說明同步，那麼直接返回true。最後上層函數會調用f.accept(s)同步執行任務。
如果異步線程不為null，那麼使用異步線程去執行this。
this的run任務如下。也就是在異步線程同步調用tryFire方法。達到其被異步線程執行的目的。
public final void run(){ tryFire(ASYNC); }其實就是先判斷源任務是否完成，如果完成，直接在對應線程執行以來任務（如果是同步，則在當前線程處理，否則在異步線程處理）
如果任務沒有完成，直接返回，因為等任務完成之後會通過postComplete去觸發調用依賴任務。
postComplete方法final void postComplete() { /* * On each step, variable f holds current dependents to pop * and run. It is extended along only one path at a time, * pushing others to avoid unbounded recursion. */ CompletableFuture<?> f = this; Completion h; while ((h = f.stack) != null || (f != this && (h = (f = this).stack) != null)) { CompletableFuture<?> d; Completion t; if (f.casStack(h, t = h.next)) { if (t != null) { if (f != this) { pushStack(h); continue; } h.next = null; // detach } f = (d = h.tryFire(NESTED)) == null ? this : d; } } }
其實邏輯很簡單，就是迭代堆棧的依賴任務。調用h.tryFire方法。NESTED就是為了避免遞歸死循環。因為FirePost會調用postComplete。如果是NESTED，則不調用。
堆棧的內容其實就是在依賴任務創建的時候加入進去的。上面我們已經提到過。
基本上述源碼已經分析了邏輯。
因為涉及異步等操作，我們需要理一下（這裡針對全異步任務）：
創建CompletableFuture成功之後會通過異步線程去執行對應任務。
如果CompletableFuture還有依賴任務（異步），會將任務加入到CompletableFuture的堆棧保存起來。以供後續完成後執行依賴任務。
當然，創建依賴任務並不只是將其加入堆棧。如果源任務在創建依賴任務的時候已經執行完成，那麼當前線程會觸發依賴任務的異步線程直接處理依賴任務。並且會告訴堆棧其他的依賴任務源任務已經完成。
主要是考慮代碼的復用。所以邏輯相對難理解。
postComplete方法會被源任務線程執行完源任務後調用。同樣也可能被依賴任務線程後調用。
執行依賴任務的方法主要就是靠tryFire方法。因為這個方法可能會被多種不同類型線程觸發，所以邏輯也繞一點。（其他依賴任務線程、源任務線程、當前依賴任務線程）
如果是當前依賴任務線程，那麼會執行依賴任務，並且會通知其他依賴任務。
如果是源任務線程，和其他依賴任務線程，則將任務轉換給依賴線程去執行。不需要通知其他依賴任務，避免死遞歸。
不得不說Doug Lea的編碼，真的是藝術。代碼的復用性全體現在邏輯上了。
轉自：大遠哥
鏈接：https://blog.csdn.net/weixin_39332800/article/details/108185931
- EOF -
按 CompletableFuture 完成順序實現 Streaming Future
你可能不知道但卻很有用的 Java 特性
優雅整潔的 Java 代碼命名技巧，風之極·淨化
看完本文有收穫？請轉發分享給更多人
關注「ImportNew」，提升Java技能
點讚和在看就是最大的支持❤️
留言列表