1.概述
CompletableFuture是jdk1.8引入的實作類，擴展了Future和CompletionStage，是一個可以在任務完成階段觸發一些操作Future，簡單的來講就是可以實作異步回呼，
2.為什么引入CompletableFuture
對于jdk1.5的Future，雖然提供了異步處理任務的能力，但是獲取結果的方式很不優雅，還是需要通過阻塞（或者輪訓）的方式，如何避免阻塞呢？其實就是注冊回呼，
業界結合觀察者模式實作異步回呼，也就是當任務執行完成后去通知觀察者，比如Netty的ChannelFuture，可以通過注冊監聽實作異步結果的處理，
Netty的ChannelFuture
public 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，引入函式式編程，通過回呼的方式去處理結果，
3.功能
CompletableFuture的功能主要體現在他的CompletionStage，
可以實作如下等功能
- 轉換（thenCompose）
- 組合（thenCombine）
- 消費（thenAccept）
- 運行（thenRun），
- 帶回傳的消費（thenApply）
消費和運行的區別：
消費使用執行結果，運行則只是運行特定任務，具體其他功能大家可以根據需求自行查看，
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，
- 如果這些依賴任務都是異步執行，那么他會通過異步執行緒池去執行任務，不能保證任務的執行順序，
上面的結論是通過閱讀源代碼得到的，下面我們深入源代碼，
4.原始碼追蹤
創建CompletableFuture
創建的方法有很多，甚至可以直接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#run
public 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 = https://www.cnblogs.com/javastack/p/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;
}
claim方法是邏輯：
- 如果異步執行緒為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，則不呼叫，
堆疊的內容其實就是在依賴任務創建的時候加入進去的，上面我們已經提到過，
4.總結
基本上述原始碼已經分析了邏輯，
因為涉及異步等操作，我們需要理一下（這里針對全異步任務）：
- 創建CompletableFuture成功之后會通過異步執行緒去執行對應任務，
- 如果CompletableFuture還有依賴任務（異步），會將任務加入到CompletableFuture的堆疊保存起來，以供后續完成后執行依賴任務，
當然，創建依賴任務并不只是將其加入堆疊，如果源任務在創建依賴任務的時候已經執行完成，那么當前執行緒會觸發依賴任務的異步執行緒直接處理依賴任務，并且會告訴堆疊其他的依賴任務源任務已經完成，
主要是考慮代碼的復用，所以邏輯相對難理解，
postComplete方法會被源任務執行緒執行完源任務后呼叫，同樣也可能被依賴任務執行緒后呼叫，
執行依賴任務的方法主要就是靠tryFire方法，因為這個方法可能會被多種不同型別執行緒觸發，所以邏輯也繞一點，（其他依賴任務執行緒、源任務執行緒、當前依賴任務執行緒）
- 如果是當前依賴任務執行緒，那么會執行依賴任務，并且會通知其他依賴任務，
- 如果是源任務執行緒，和其他依賴任務執行緒，則將任務轉換給依賴執行緒去執行，不需要通知其他依賴任務，避免死遞回，
不得不說Doug Lea的編碼，真的是藝術，代碼的復用性全體現在邏輯上了，
來源：blog.csdn.net/weixin_39332800/article/details/108185931
近期熱文推薦：
1.1,000+ 道 Java面試題及答案整理(2022最新版)
2.勁爆！Java 協程要來了，，，
3.Spring Boot 2.x 教程，太全了！
4.別再寫滿屏的爆爆爆炸類了，試試裝飾器模式，這才是優雅的方式！！
5.《Java開發手冊（嵩山版）》最新發布，速速下載！
覺得不錯，別忘了隨手點贊+轉發哦！
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/452816.html
標籤：Java
上一篇：團隊vue基礎鏡像選擇思考
下一篇：Java 將CSV轉為Excel