Java8中CompletableFuture使用場景與實現(xiàn)原理
1.概述
CompletableFuture是jdk1.8引入的實現(xiàn)類。擴展了Future和CompletionStage，是一個可以在任務完成階段觸發(fā)一些操作Future。簡單的來講就是可以實現(xiàn)異步回調。
2.為什么引入CompletableFuture
對于jdk1.5的Future，雖然提供了異步處理任務的能力，但是獲取結果的方式很不優(yōu)雅，還是需要通過阻塞（或者輪訓）的方式。如何避免阻塞呢？其實就是注冊回調。
業(yè)界結合觀察者模式實現(xiàn)異步回調。也就是當任務執(zhí)行完成后去通知觀察者。比如Netty的ChannelFuture，可以通過注冊監(jiān)聽實現(xiàn)異步結果的處理。
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方法注冊監(jiān)聽。如果任務完成，會調用notifyListeners通知。
CompletableFuture通過擴展Future，引入函數式編程，通過回調的方式去處理結果。
3.功能
CompletableFuture的功能主要體現(xiàn)在他的CompletionStage。
可以實現(xiàn)如下等功能
- 轉換（thenCompose）
- 組合（thenCombine）
- 消費（thenAccept）
- 運行（thenRun）。
- 帶返回的消費（thenApply）
消費和運行的區(qū)別：
消費使用執(zhí)行結果。運行則只是運行特定任務。具體其他功能大家可以根據需求自行查看。
CompletableFuture借助CompletionStage的方法可以實現(xiàn)鏈式調用。并且可以選擇同步或者異步兩種方式。
這里舉個簡單的例子來體驗一下他的功能。
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");
});
}
執(zhí)行結果
supplyAsync pool-1-thread-1
helloworld
ddddd
ddddsd
Thread[main,5,main]
dddaewdd
根據結果我們可以看到會有序執(zhí)行對應任務。
注意：
如果是同步執(zhí)行cf.thenRun。他的執(zhí)行線程可能main線程，也可能是執(zhí)行源任務的線程。如果執(zhí)行源任務的線程在main調用之前執(zhí)行完了任務。那么cf.thenRun方法會由main線程調用。
這里說明一下，如果是同一任務的依賴任務有多個：
- 如果這些依賴任務都是同步執(zhí)行。那么假如這些任務被當前調用線程（main）執(zhí)行，則是有序執(zhí)行，假如被執(zhí)行源任務的線程執(zhí)行，那么會是倒序執(zhí)行。因為內部任務數據結構為LIFO。
- 如果這些依賴任務都是異步執(zhí)行，那么他會通過異步線程池去執(zhí)行任務。不能保證任務的執(zhí)行順序。
上面的結論是通過閱讀源代碼得到的。下面我們深入源代碼。
3.源碼追蹤
創(chuàng)建CompletableFuture
創(chuàng)建的方法有很多，甚至可以直接new一個。我們來看一下supplyAsync異步創(chuàng)建的方法。
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，帶返回值的函數。如果是異步方法，并且傳遞了執(zhí)行器，那么會使用傳入的執(zhí)行器去執(zhí)行任務。否則采用公共的ForkJoin并行線程池，如果不支持并行，新建一個線程去執(zhí)行。
這里我們需要注意ForkJoin是通過守護線程去執(zhí)行任務的。所以必須有非守護線程的存在才行。
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;
}
這里會創(chuàng)建一個用于返回的CompletableFuture。
然后構造一個AsyncSupply，并將創(chuàng)建的CompletableFuture作為構造參數傳入。
那么，任務的執(zhí)行完全依賴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();
}
}
1.該方法會調用Supplier的get方法。并將結果設置到CompletableFuture中。我們應該清楚這些操作都是在異步線程中調用的。
2.d.postComplete方法就是通知任務執(zhí)行完成。觸發(fā)后續(xù)依賴任務的執(zhí)行，也就是實現(xiàn)CompletionStage的關鍵點。
在看postComplete方法之前我們先來看一下創(chuàng)建依賴任務的邏輯。
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邏輯：
1.構造一個CompletableFuture，主要是為了鏈式調用。
2.如果為異步任務，直接返回。因為源任務結束后會觸發(fā)異步線程執(zhí)行對應邏輯。
3.如果為同步任務（e==null），會調用d.uniAccept方法。這個方法在這里邏輯：如果源任務完成，調用f，返回true。否則進入if代碼塊（Mark 1）。
4.如果是異步任務直接進入if（Mark 1）。
Mark1邏輯：
1.構造一個UniAccept，將其push入棧。這里通過CAS實現(xiàn)樂觀鎖實現(xiàn)。
2.調用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);
}
1.會調用d.uniAccept方法。其實該方法判斷源任務是否完成，如果完成則執(zhí)行依賴任務，否則返回false。
2.如果依賴任務已經執(zhí)行，調用d.postFire，主要就是Fire的后續(xù)處理。根據不同模式邏輯不同。
這里簡單說一下，其實mode有同步異步，和迭代。迭代為了避免無限遞歸。
這里強調一下d.uniAccept方法的第三個參數。
如果是異步調用（mode>0），傳入null。否則傳入this。
區(qū)別看下面代碼。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)同步執(zhí)行任務。
- 如果異步線程不為null，那么使用異步線程去執(zhí)行this。
this的run任務如下。也就是在異步線程同步調用tryFire方法。達到其被異步線程執(zhí)行的目的。
public final void run() { tryFire(ASYNC); }
看完上面的邏輯，我們基本理解依賴任務的邏輯。
其實就是先判斷源任務是否完成，如果完成，直接在對應線程執(zhí)行以來任務（如果是同步，則在當前線程處理，否則在異步線程處理）
如果任務沒有完成，直接返回，因為等任務完成之后會通過postComplete去觸發(fā)調用依賴任務。
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就是為了避免遞歸死循環(huán)。因為FirePost會調用postComplete。如果是NESTED，則不調用。
堆棧的內容其實就是在依賴任務創(chuàng)建的時候加入進去的。上面我們已經提到過。
4.總結
基本上述源碼已經分析了邏輯。
因為涉及異步等操作，我們需要理一下（這里針對全異步任務）：
1.創(chuàng)建CompletableFuture成功之后會通過異步線程去執(zhí)行對應任務。
2.如果CompletableFuture還有依賴任務（異步），會將任務加入到CompletableFuture的堆棧保存起來。以供后續(xù)完成后執(zhí)行依賴任務。
當然，創(chuàng)建依賴任務并不只是將其加入堆棧。如果源任務在創(chuàng)建依賴任務的時候已經執(zhí)行完成，那么當前線程會觸發(fā)依賴任務的異步線程直接處理依賴任務。并且會告訴堆棧其他的依賴任務源任務已經完成。
主要是考慮代碼的復用。所以邏輯相對難理解。
postComplete方法會被源任務線程執(zhí)行完源任務后調用。同樣也可能被依賴任務線程后調用。
執(zhí)行依賴任務的方法主要就是靠tryFire方法。因為這個方法可能會被多種不同類型線程觸發(fā)，所以邏輯也繞一點。（其他依賴任務線程、源任務線程、當前依賴任務線程）
- 如果是當前依賴任務線程，那么會執(zhí)行依賴任務，并且會通知其他依賴任務。
- 如果是源任務線程，和其他依賴任務線程，則將任務轉換給依賴線程去執(zhí)行。不需要通知其他依賴任務，避免死遞歸。
不得不說Doug Lea的編碼，真的是藝術。代碼的復用性全體現(xiàn)在邏輯上了。
到此這篇關于Java8中CompletableFuture使用場景與實現(xiàn)原理的文章就介紹到這了,更多相關CompletableFuture使用場景與原理內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
本文主要介紹了springboot 按月分表的實現(xiàn)方式，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-04-04
這篇文章主要大家介紹了關于如何在Java中優(yōu)雅地判空的相關資料，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2018-11-11
這篇文章主要介紹了Spring MVC的參數綁定和返回值問題，本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2021-02-02
這篇文章主要介紹了分享JVM 的四種引用方式，文章圍繞主題展開詳細的內容介紹，具有一定的參考價值，需要的小伙伴可以參考一下2022-07-07
Spring?Boot實現(xiàn)web.xml功能示例詳解
這篇文章主要介紹了Spring?Boot實現(xiàn)web.xml功能,通過本文介紹我們了解到，在Spring Boot應用中，我們可以通過注解和編程兩種方式實現(xiàn)web.xml的功能，包括如何創(chuàng)建及注冊Servlet、Filter以及Listener等，需要的朋友可以參考下2023-09-09
這篇文章主要介紹了淺談SpringCloud之Ribbon,文中有非常詳細的代碼示例,對正在學習SpringCloud的小伙伴們有很大的幫助,需要的朋友可以參考下2021-05-05
這篇文章主要介紹了RocketMQ中的通信模塊詳解,RocketMQ消息隊列集群主要包括NameServer、Broker(Master/Slave)、Producer、Consumer4個角色,本文我們簡單來講解一下,需要的朋友可以參考下2024-01-01
排序就是將一串記錄按照其中某個或某些關鍵字的大小，遞增或遞減的排列起來的操作。本文將用Java實現(xiàn)一些基本的排序算法，感興趣的可以了解一下2022-07-07
mybatis的insert語句插入數據時的返回值的實現(xiàn)
這篇文章主要介紹了mybatis的insert語句插入數據時的返回值的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-10-10
上個月，idea的使用量超越eclipse的消息席卷了整個IT界，idea到底好在哪里呢?下面小編通過本文給大家詳細介紹下IDEA入門級使用教程，非常詳細，感興趣的朋友一起看看吧2020-10-10