Java CompletableFuture 異步超時(shí)實(shí)現(xiàn)深入研究
前言
作者:京東科技 張?zhí)熨n
JDK 8 是一次重大的版本升級(jí),新增了非常多的特性,其中之一便是 CompletableFuture
。自此從 JDK 層面真正意義上的支持了基于事件的異步編程范式,彌補(bǔ)了 Future
的缺陷。
在我們的日常優(yōu)化中,最常用手段便是多線程并行執(zhí)行。這時(shí)候就會(huì)涉及到 CompletableFuture
的使用。
常見使用方式
下面舉例一個(gè)常見場景。
假如我們有兩個(gè) RPC 遠(yuǎn)程調(diào)用服務(wù),我們需要獲取兩個(gè) RPC 的結(jié)果后,再進(jìn)行后續(xù)邏輯處理。
public static void main(String[] args) { // 任務(wù) A,耗時(shí) 2 秒 int resultA = compute(1); // 任務(wù) B,耗時(shí) 2 秒 int resultB = compute(2); // 后續(xù)業(yè)務(wù)邏輯處理 System.out.println(resultA + resultB); }
可以預(yù)估到,串行執(zhí)行最少耗時(shí) 4 秒,并且 B 任務(wù)并不依賴 A 任務(wù)結(jié)果。
對于這種場景,我們通常會(huì)選擇并行的方式優(yōu)化,Demo 代碼如下:
public static void main(String[] args) { // 僅簡單舉例,在生產(chǎn)代碼中可別這么寫! // 統(tǒng)計(jì)耗時(shí)的函數(shù) time(() -> { CompletableFuture<Integer> result = Stream.of(1, 2) // 創(chuàng)建異步任務(wù) .map(x -> CompletableFuture.supplyAsync(() -> compute(x), executor)) // 聚合 .reduce(CompletableFuture.completedFuture(0), (x, y) -> x.thenCombineAsync(y, Integer::sum, executor)); // 等待結(jié)果 try { System.out.println("結(jié)果:" + result.get()); } catch (ExecutionException | InterruptedException e) { System.err.println("任務(wù)執(zhí)行異常"); } }); } 輸出: [async-1]: 任務(wù)執(zhí)行開始:1 [async-2]: 任務(wù)執(zhí)行開始:2 [async-1]: 任務(wù)執(zhí)行完成:1 [async-2]: 任務(wù)執(zhí)行完成:2 結(jié)果:3 耗時(shí):2 秒
可以看到耗時(shí)變成了 2 秒。
存在的問題
分析
看上去 CompletableFuture
現(xiàn)有功能可以滿足我們訴求。但當(dāng)我們引入一些現(xiàn)實(shí)常見情況時(shí),一些潛在的不足便暴露出來了。
compute(x)
如果是一個(gè)根據(jù)入?yún)⒉樵冇脩裟愁愋蛢?yōu)惠券列表的任務(wù),我們需要查詢兩種優(yōu)惠券并組合在一起返回給上游。假如上游要求我們 2 秒內(nèi)處理完畢并返回結(jié)果,但 compute(x)
耗時(shí)卻在 0.5 秒 ~ 無窮大波動(dòng)。這時(shí)候我們就需要把耗時(shí)過長的 compute(x)
任務(wù)結(jié)果放棄,僅處理在指定時(shí)間內(nèi)完成的任務(wù),盡可能保證服務(wù)可用。
那么以上代碼的耗時(shí)由耗時(shí)最長的服務(wù)決定,無法滿足現(xiàn)有訴求。通常我們會(huì)使用 get(long timeout, TimeUnit unit)
來指定獲取結(jié)果的超時(shí)時(shí)間,并且我們會(huì)給 compute(x)
設(shè)置一個(gè)超時(shí)時(shí)間,達(dá)到后自動(dòng)拋異常來中斷任務(wù)。
public static void main(String[] args) { // 僅簡單舉例,在生產(chǎn)代碼中可別這么寫! // 統(tǒng)計(jì)耗時(shí)的函數(shù) time(() -> { List<CompletableFuture<Integer>> result = Stream.of(1, 2) // 創(chuàng)建異步任務(wù),compute(x) 超時(shí)拋出異常 .map(x -> CompletableFuture.supplyAsync(() -> compute(x), executor)) .toList(); // 等待結(jié)果 int res = 0; for (CompletableFuture<Integer> future : result) { try { res += future.get(2, SECONDS); } catch (ExecutionException | InterruptedException | TimeoutException e) { System.err.println("任務(wù)執(zhí)行異?;虺瑫r(shí)"); } } System.out.println("結(jié)果:" + res); }); } 輸出: [async-2]: 任務(wù)執(zhí)行開始:2 [async-1]: 任務(wù)執(zhí)行開始:1 [async-1]: 任務(wù)執(zhí)行完成:1 任務(wù)執(zhí)行異?;虺瑫r(shí) 結(jié)果:1 耗時(shí):2 秒
可以看到,只要我們能夠給 compute(x)
設(shè)置一個(gè)超時(shí)時(shí)間將任務(wù)中斷,結(jié)合 get
、getNow
等獲取結(jié)果的方式,就可以很好地管理整體耗時(shí)。
那么問題也就轉(zhuǎn)變成了,如何給任務(wù)設(shè)置異步超時(shí)時(shí)間呢?
現(xiàn)有做法
當(dāng)異步任務(wù)是一個(gè) RPC 請求時(shí),我們可以設(shè)置一個(gè) JSF 超時(shí),以達(dá)到異步超時(shí)效果。
當(dāng)請求是一個(gè) R2M 請求時(shí),我們也可以控制 R2M 連接的最大超時(shí)時(shí)間來達(dá)到效果。
這么看好像我們都是在依賴三方中間件的能力來管理任務(wù)超時(shí)時(shí)間?那么就存在一個(gè)問題,中間件超時(shí)控制能力有限,如果異步任務(wù)是中間件 IO 操作 + 本地計(jì)算操作怎么辦?
用 JSF 超時(shí)舉一個(gè)具體的例子,反編譯 JSF 的獲取結(jié)果代碼如下:
public V get(long timeout, TimeUnit unit) throws InterruptedException { // 配置的超時(shí)時(shí)間 timeout = unit.toMillis(timeout); // 剩余等待時(shí)間 long remaintime = timeout - (this.sentTime - this.genTime); if (remaintime <= 0L) { if (this.isDone()) { // 反序列化獲取結(jié)果 return this.getNow(); } } else if (this.await(remaintime, TimeUnit.MILLISECONDS)) { // 等待時(shí)間內(nèi)任務(wù)完成,反序列化獲取結(jié)果 return this.getNow(); } this.setDoneTime(); // 超時(shí)拋出異常 throw this.clientTimeoutException(false); }
當(dāng)這個(gè)任務(wù)剛好卡在超時(shí)邊緣完成時(shí),這個(gè)任務(wù)的耗時(shí)時(shí)間就變成了超時(shí)時(shí)間 + 獲取結(jié)果時(shí)間。而獲取結(jié)果(反序列化)作為純本地計(jì)算操作,耗時(shí)長短受 CPU 影響較大。
某些 CPU 使用率高的情況下,就會(huì)出現(xiàn)異步任務(wù)沒能觸發(fā)拋出異常中斷,導(dǎo)致我們無法準(zhǔn)確控制超時(shí)時(shí)間。對上游來說,本次請求全部失敗。
解決方式
JDK 9
這類問題非常常見,如大促場景,服務(wù)器 CPU 瞬間升高就會(huì)出現(xiàn)以上問題。
那么如何解決呢?其實(shí) JDK 的開發(fā)大佬們早有研究。在 JDK 9,CompletableFuture
正式提供了 orTimeout
、completeTimeout
方法,來準(zhǔn)確實(shí)現(xiàn)異步超時(shí)控制。
public CompletableFuture<T> orTimeout(long timeout, TimeUnit unit) { if (unit == null) throw new NullPointerException(); if (result == null) whenComplete(new Canceller(Delayer.delay(new Timeout(this), timeout, unit))); return this; }
JDK 9 orTimeout
其實(shí)現(xiàn)原理是通過一個(gè)定時(shí)任務(wù),在給定時(shí)間之后拋出異常。如果任務(wù)在指定時(shí)間內(nèi)完成,則取消拋異常的操作。
以上代碼我們按執(zhí)行順序來看下:
首先執(zhí)行 new Timeout(this)
。
static final class Timeout implements Runnable { final CompletableFuture<?> f; Timeout(CompletableFuture<?> f) { this.f = f; } public void run() { if (f != null && !f.isDone()) // 拋出超時(shí)異常 f.completeExceptionally(new TimeoutException()); } }
通過源碼可以看到,Timeout
是一個(gè)實(shí)現(xiàn) Runnable 的類,run()
方法負(fù)責(zé)給傳入的異步任務(wù)通過 completeExceptionally
CAS 賦值異常,將任務(wù)標(biāo)記為異常完成。
那么誰來觸發(fā)這個(gè) run()
方法呢?我們看下 Delayer
的實(shí)現(xiàn)。
static final class Delayer { static ScheduledFuture<?> delay(Runnable command, long delay, TimeUnit unit) { // 到時(shí)間觸發(fā) command 任務(wù) return delayer.schedule(command, delay, unit); } static final class DaemonThreadFactory implements ThreadFactory { public Thread newThread(Runnable r) { Thread t = new Thread(r); t.setDaemon(true); t.setName("CompletableFutureDelayScheduler"); return t; } } static final ScheduledThreadPoolExecutor delayer; static { (delayer = new ScheduledThreadPoolExecutor( 1, new DaemonThreadFactory())). setRemoveOnCancelPolicy(true); } }
Delayer
其實(shí)就是一個(gè)單例定時(shí)調(diào)度器,Delayer.delay(new Timeout(this), timeout, unit)
通過 ScheduledThreadPoolExecutor
實(shí)現(xiàn)指定時(shí)間后觸發(fā) Timeout
的 run()
方法。
到這里就已經(jīng)實(shí)現(xiàn)了超時(shí)拋出異常的操作。但當(dāng)任務(wù)完成時(shí),就沒必要觸發(fā) Timeout
了。因此我們還需要實(shí)現(xiàn)一個(gè)取消邏輯。
static final class Canceller implements BiConsumer<Object, Throwable> { final Future<?> f; Canceller(Future<?> f) { this.f = f; } public void accept(Object ignore, Throwable ex) { if (ex == null && f != null && !f.isDone()) // 3 未觸發(fā)拋異常任務(wù)則取消 f.cancel(false); } }
當(dāng)任務(wù)執(zhí)行完成,或者任務(wù)執(zhí)行異常時(shí),我們也就沒必要拋出超時(shí)異常了。因此我們可以把 delayer.schedule(command, delay, unit)
返回的定時(shí)超時(shí)任務(wù)取消,不再觸發(fā) Timeout
。 當(dāng)我們的異步任務(wù)完成,并且定時(shí)超時(shí)任務(wù)未完成的時(shí)候,就是我們?nèi)∠臅r(shí)機(jī)。因此我們可以通過 whenComplete(BiConsumer<? super T, ? super Throwable> action)
來完成。
Canceller
就是一個(gè) BiConsumer
的實(shí)現(xiàn)。其持有了 delayer.schedule(command, delay, unit)
返回的定時(shí)超時(shí)任務(wù),accept(Object ignore, Throwable ex)
實(shí)現(xiàn)了定時(shí)超時(shí)任務(wù)未完成后,執(zhí)行 cancel(boolean mayInterruptIfRunning)
取消任務(wù)的操作。
JDK 8
如果我們使用的是 JDK 9 或以上,我們可以直接用 JDK 的實(shí)現(xiàn)來完成異步超時(shí)操作。那么 JDK 8 怎么辦呢?
其實(shí)我們也可以根據(jù)上述邏輯簡單實(shí)現(xiàn)一個(gè)工具類來輔助。
以下是我們營銷自己的工具類以及用法,貼出來給大家作為參考,大家也可以自己寫的更優(yōu)雅一些~
調(diào)用方式:
CompletableFutureExpandUtils.orTimeout(異步任務(wù), 超時(shí)時(shí)間, 時(shí)間單位);
工具類源碼:
package com.jd.jr.market.reduction.util; import com.jdpay.market.common.exception.UncheckedException; import java.util.concurrent.*; import java.util.function.BiConsumer; /** * CompletableFuture 擴(kuò)展工具 * * @author zhangtianci7 */ public class CompletableFutureExpandUtils { /** * 如果在給定超時(shí)之前未完成,則異常完成此 CompletableFuture 并拋出 {@link TimeoutException} 。 * * @param timeout 在出現(xiàn) TimeoutException 異常完成之前等待多長時(shí)間,以 {@code unit} 為單位 * @param unit 一個(gè) {@link TimeUnit},結(jié)合 {@code timeout} 參數(shù),表示給定粒度單位的持續(xù)時(shí)間 * @return 入?yún)⒌?CompletableFuture */ public static <T> CompletableFuture<T> orTimeout(CompletableFuture<T> future, long timeout, TimeUnit unit) { if (null == unit) { throw new UncheckedException("時(shí)間的給定粒度不能為空"); } if (null == future) { throw new UncheckedException("異步任務(wù)不能為空"); } if (future.isDone()) { return future; } return future.whenComplete(new Canceller(Delayer.delay(new Timeout(future), timeout, unit))); } /** * 超時(shí)時(shí)異常完成的操作 */ static final class Timeout implements Runnable { final CompletableFuture<?> future; Timeout(CompletableFuture<?> future) { this.future = future; } public void run() { if (null != future && !future.isDone()) { future.completeExceptionally(new TimeoutException()); } } } /** * 取消不需要的超時(shí)的操作 */ static final class Canceller implements BiConsumer<Object, Throwable> { final Future<?> future; Canceller(Future<?> future) { this.future = future; } public void accept(Object ignore, Throwable ex) { if (null == ex && null != future && !future.isDone()) { future.cancel(false); } } } /** * 單例延遲調(diào)度器,僅用于啟動(dòng)和取消任務(wù),一個(gè)線程就足夠 */ static final class Delayer { static ScheduledFuture<?> delay(Runnable command, long delay, TimeUnit unit) { return delayer.schedule(command, delay, unit); } static final class DaemonThreadFactory implements ThreadFactory { public Thread newThread(Runnable r) { Thread t = new Thread(r); t.setDaemon(true); t.setName("CompletableFutureExpandUtilsDelayScheduler"); return t; } } static final ScheduledThreadPoolExecutor delayer; static { delayer = new ScheduledThreadPoolExecutor(1, new DaemonThreadFactory()); delayer.setRemoveOnCancelPolicy(true); } } }
參考資料 JEP 266: JDK 9 并發(fā)包更新提案
以上就是Java CompletableFuture 異步超時(shí)實(shí)現(xiàn)深入研究的詳細(xì)內(nèi)容,更多關(guān)于Java CompletableFuture 異步超時(shí)的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
Spring整合CXF webservice restful實(shí)例詳解
這篇文章主要為大家詳細(xì)介紹了Spring整合CXF webservice restful的實(shí)例,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2017-08-08Spring異常實(shí)現(xiàn)統(tǒng)一處理的方法
這篇文章主要介紹了Spring異常實(shí)現(xiàn)統(tǒng)一處理的方法,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)吧2022-12-12Spring?中?PageHelper?不生效問題及解決方法
這篇文章主要介紹了Spring?中?PageHelper?不生效問題,使用這個(gè)插件時(shí)要注意版本的問題,不同的版本可能 PageHelper 不會(huì)生效,本文結(jié)合示例代碼給大家介紹的非常詳細(xì),需要的朋友可以參考下2022-12-12使用springboot單元測試對weblistener的加載測試
這篇文章主要介紹了使用springboot單元測試對weblistener的加載測試,具有很好的參考價(jià)值,希望對大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2021-10-10springboot2.0+elasticsearch5.5+rabbitmq搭建搜索服務(wù)的坑
這篇文章主要介紹了springboot2.0+elasticsearch5.5+rabbitmq搭建搜索服務(wù)的坑,小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過來看看吧2018-06-06Spring Cloud Gateway 使用JWT工具類做用戶登錄校驗(yàn)功能
這篇文章主要介紹了Spring Cloud Gateway 使用JWT工具類做用戶登錄校驗(yàn)的示例代碼,本文給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2021-01-01