Java中CompletableFuture四種調(diào)用模式的實(shí)現(xiàn)
三種調(diào)用模式
CompletableFuture(以下簡(jiǎn)稱(chēng)CF)提供了三種調(diào)用模式,分別是就地執(zhí)行、異步使用默認(rèn)執(zhí)行器執(zhí)行、異步指定執(zhí)行器執(zhí)行。
就地執(zhí)行指的是回調(diào)在當(dāng)前線(xiàn)程中執(zhí)行,調(diào)用thenApply、thenCompose等方法時(shí),如果當(dāng)前 CF 已經(jīng)執(zhí)行完成,會(huì)立即執(zhí)行回調(diào),稱(chēng)為當(dāng)前執(zhí)行,執(zhí)行時(shí)刻為當(dāng)前(now);如果未完成,則由完成 CF 的線(xiàn)程執(zhí)行。
如下分別是立即執(zhí)行和異步執(zhí)行的例子:
var cf = CompletableFuture.completed(1); cf.thenApply(x -> x + 1) .thenRun(x -> System.out.println(x)) .join();
以上代碼全程同步。
var cf = new CompletableFuture<Integer>(); cf.thenApply(x -> x + 1) .thenRun(x -> System.out.println(x)); new Thread(() -> cf.complete(1)).start(); Uninterruptible.sleep(1, TimeUnit.SECONDS);
thenApply、thenRun在調(diào)用時(shí),cf未完成,無(wú)法立刻執(zhí)行,其執(zhí)行在完成cf的線(xiàn)程,也就是新創(chuàng)建的線(xiàn)程中。
異步執(zhí)行指的是回調(diào)任務(wù)的執(zhí)行必定在執(zhí)行器中執(zhí)行,默認(rèn)執(zhí)行器為Java提供的commonPool線(xiàn)程池,當(dāng)然也可以通過(guò)重寫(xiě)defaultExecutor實(shí)現(xiàn)調(diào)用指定的線(xiàn)程池。
var cf = CompletableFuture.completed(1); cf.thenApplyAsync(x -> x + 1) .thenRunAsync(x -> System.out.println(x)) .join(); Uninterruptible.sleep(1, TimeUnit.SECONDS);
以上代碼中打印操作在公共線(xiàn)程池中執(zhí)行。
比較
就地執(zhí)行性能最好,可以完全避免線(xiàn)程上下文切換,適合執(zhí)行一些輕量級(jí)任務(wù)。缺點(diǎn)是使用不當(dāng)時(shí),會(huì)阻塞當(dāng)前線(xiàn)程;可能會(huì)造成“線(xiàn)程泄露”,導(dǎo)致線(xiàn)程池中的線(xiàn)程沒(méi)有及時(shí)歸還。
異步執(zhí)行反之。
第 4 種調(diào)用模式
線(xiàn)程池中任務(wù)執(zhí)行有一條原則:盡最大努力交付。意思是如果任務(wù)提交時(shí)沒(méi)有拒絕,沒(méi)有拋出拒絕執(zhí)行等異常,通常來(lái)說(shuō)通過(guò)了信號(hào)量、限流器、執(zhí)行時(shí)間、線(xiàn)程數(shù)等諸多限制,后續(xù)的執(zhí)行應(yīng)該不作額外限制,且努力完成;而不是等執(zhí)行過(guò)程中再拋出類(lèi)似拒絕服務(wù)等異常。反過(guò)來(lái)說(shuō),如果當(dāng)前任務(wù)提交時(shí),任務(wù)不能執(zhí)行,就應(yīng)該拒絕執(zhí)行。這條簡(jiǎn)單的原則可以避免考慮復(fù)雜的問(wèn)題,比如反壓、取消機(jī)制等,也能夠應(yīng)對(duì)大多數(shù)的業(yè)務(wù)場(chǎng)景。
對(duì)于非輕量級(jí)任務(wù),例如 A -> B,表示任務(wù)A執(zhí)行完成后執(zhí)行任務(wù)B,常規(guī)的線(xiàn)程池實(shí)現(xiàn)有一個(gè)問(wèn)題,B任務(wù)的提交不一定立即執(zhí)行,可能遇到排隊(duì)(進(jìn)入阻塞隊(duì)列)甚至超時(shí)等情況,最終導(dǎo)致整個(gè)任務(wù)的滯后。此時(shí)如果能就地執(zhí)行最好。
如果選擇就地執(zhí)行策略,解決了以上問(wèn)題,但是可能會(huì)導(dǎo)致CF已完成后執(zhí)行的當(dāng)前線(xiàn)程阻塞。這時(shí)最好有執(zhí)行器執(zhí)行任務(wù),而不是占用當(dāng)前線(xiàn)程。
最近CFFU類(lèi)庫(kù)提供LLCF#relayAsync0,完美解決了以上痛點(diǎn)。LL表示low level,對(duì)于其的正確使用要求開(kāi)發(fā)人員對(duì)CompletableFuture有著充分的理解。relay的含義是接力,這里指的是
- relay Async 接力異步
- Async 詞尾,保證一定是異步(和CF命名表義 一樣)
異步時(shí)(不阻塞調(diào)用邏輯),用前個(gè)computation的線(xiàn)程接力執(zhí)行,不使用新線(xiàn)程,避免了上下文切換開(kāi)銷(xiāo)。
例子
relayAsync0 簽名如下:
public static <T, F extends CompletionStage<?>> F relayAsync0(
CompletionStage<? extends T> cfThis,
Function<CompletableFuture<T>, F> relayComputations, Executor executor)
需要注意傳入的回調(diào)任務(wù)不是普通的Function,而是入?yún)F,出參 CompletionStage,也就是說(shuō)我們需要傳入對(duì)CF的回調(diào)。比如:
cf -> cf.thenApply(...) cf -> cf.thenCompose(...) cf -> cf.thenRun(...)
該方法使用時(shí)和thenApplyAsync很像,只不過(guò)由實(shí)例方法調(diào)用改成了靜態(tài)方法調(diào)用,回調(diào)參數(shù)為對(duì)CF的回調(diào)。
以下代碼引用自CFFU作者 李鼎 | Jerry Lee,詳細(xì)說(shuō)明四種調(diào)用模式的用法:
public class RelayAsyncDescriptionAndExample {
static void executeComputationsOfNewStage(CompletableFuture<String> cf) {
// ================================================================================
// Default execution
// ================================================================================
cf.thenApply(s -> {
// a simulating long-running computation...
sleep(1000);
// if input cf is COMPLETED when computations execute,
// executes the long time computation SYNCHRONOUSLY (aka. in the caller thread);
// this SYNCHRONIZED execution leads to BLOCKing sequential codes of caller... ??
return s + s;
});
// ================================================================================
// Asynchronous execution of CompletableFuture(default executor or custom executor)
// ================================================================================
cf.thenApplyAsync(s -> {
// a simulating long-running computation...
sleep(1000);
// always executes via an executor(guarantees not to block sequential code of caller).
// if input cf is INCOMPLETE when computations execute,
// the execution via an executor leads to ONE MORE thread switching. ??
return s + s;
});
// ================================================================================
// How about the fourth way to arrange execution of a new stage's computations?
// ================================================================================
//
// - if input cf is COMPLETED when computations execute, use "asynchronous execution" (via supplied Executor),
// won't block sequential code of caller ?
// - otherwise, use "default execution", save one thread switching ?
//
// Let's call this way as "relay async".
LLCF.relayAsync0(cf, f -> f.thenApply(s -> {
// a simulating long-running computation...
sleep(1000);
// if input cf is COMPLETED, executes via supplied executor
// if input cf is INCOMPLETE, use "default execution"
return s + s;
}), ForkJoinPool.commonPool());
}
}
實(shí)現(xiàn)分析
public static <T, F extends CompletionStage<?>> F relayAsync0(
CompletionStage<? extends T> cfThis,
Function<CompletableFuture<T>, F> relayComputations, Executor executor) {
final CompletableFuture<T> promise = new CompletableFuture<>();
final F ret = relayComputations.apply(promise);
final Thread callerThread = currentThread();
final boolean[] returnedFromPeek0 = {false};
LLCF.peek0(cfThis, (v, ex) -> {
if (currentThread().equals(callerThread) && !returnedFromPeek0[0]) {
// If the action is running in the caller thread(single same thread) and `peek0` invocation does not
// return to caller(flag returnedFromPeek0 is false), the action is being executed synchronously.
// To prevent blocking the caller's sequential code, use the supplied executor to complete the promise.
executor.execute(() -> completeCf0(promise, v, ex));
} else {
// Otherwise, complete the promise directly, avoiding one thread switching.
completeCf0(promise, v, ex);
}
}, "relayAsync0");
returnedFromPeek0[0] = true;
return ret;
}
說(shuō)明:
- completeCf0方法可以將結(jié)果v或者異常ex設(shè)置到promise中
- peek0 近似等效于 whenComplete
分析:
- 可以通過(guò)引入新的CF,也就是 promise 實(shí)現(xiàn)線(xiàn)程傳遞,其他線(xiàn)程“完成”promise時(shí),這個(gè)線(xiàn)程隱式傳到了promise中,可以理解成隱式上下文。任何一個(gè)CF都帶有一個(gè)隱式上下文。
- returnedFromPeek0 避免了異步調(diào)用但是恰好是同線(xiàn)程的問(wèn)題,此時(shí)也應(yīng)該實(shí)現(xiàn)relay語(yǔ)義,因?yàn)槲覀兊哪康氖潜苊鈱?duì)當(dāng)前線(xiàn)程的阻塞。returnedFromPeek0 天然線(xiàn)程安全,因?yàn)槠湓L(fǎng)問(wèn)總是在一個(gè)確定的線(xiàn)程內(nèi)。
- else 代碼塊:就地執(zhí)行,避免線(xiàn)程切換。
總結(jié)
到此這篇關(guān)于Java中CompletableFuture四種調(diào)用模式的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)Java CompletableFuture 調(diào)用模式內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
springCloud gateWay 統(tǒng)一鑒權(quán)的實(shí)現(xiàn)代碼
這篇文章主要介紹了springCloud gateWay 統(tǒng)一鑒權(quán)的實(shí)現(xiàn)代碼,統(tǒng)一鑒權(quán)包括鑒權(quán)邏輯和代碼實(shí)現(xiàn),本文給大家介紹的非常詳細(xì),需要的朋友可以參考下2022-02-02
關(guān)于Object中equals方法和hashCode方法判斷的分析
今天小編就為大家分享一篇關(guān)于關(guān)于Object中equals方法和hashCode方法判斷的分析,小編覺(jué)得內(nèi)容挺不錯(cuò)的,現(xiàn)在分享給大家,具有很好的參考價(jià)值,需要的朋友一起跟隨小編來(lái)看看吧2019-01-01
SpringBoot修改內(nèi)置tomcat版本的操作步驟
生產(chǎn)環(huán)境使用的外部部署Tomcat還是內(nèi)置Tomcat由于版本安全漏洞,往往需要升級(jí)到指定的安全版本,本文演示一下SpringBoot升級(jí)內(nèi)置的Tomcat版本,感興趣的小伙伴跟著小編一起來(lái)看看吧2024-07-07
Mybatis many=@Many的傳值問(wèn)題解決
本文詳細(xì)介紹了在MyBatis通過(guò)主查詢(xún)將單個(gè)或多個(gè)值傳遞給子查詢(xún)的方法,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2025-11-11
Java如何判斷一個(gè)字符串是否包含某個(gè)字符串
這篇文章主要給大家介紹了關(guān)于Java如何判斷一個(gè)字符串是否包含某個(gè)字符串的相關(guān)資料,在實(shí)際編程中,經(jīng)常需要判斷一個(gè)字符串中是否包含某個(gè)子串,需要的朋友可以參考下2023-07-07

