Java CompletableFuture 異步超時(shí)實(shí)現(xiàn)探索
簡(jiǎn)介
JDK 8 中 CompletableFuture 沒有超時(shí)中斷任務(wù)的能力?，F(xiàn)有做法強(qiáng)依賴任務(wù)自身的超時(shí)實(shí)現(xiàn)。本文提出一種異步超時(shí)實(shí)現(xiàn)方案，解決上述問題。
前言
JDK 8 是一次重大的版本升級(jí)，新增了非常多的特性，其中之一便是 CompletableFuture。自此從 JDK 層面真正意義上的支持了基于事件的異步編程范式，彌補(bǔ)了 Future 的缺陷。
在我們的日常優(yōu)化中，最常用手段便是多線程并行執(zhí)行。這時(shí)候就會(huì)涉及到 CompletableFuture 的使用。
常見使用方式
下面舉例一個(gè)常見場(chǎng)景。
假如我們有兩個(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é)果。
對(duì)于這種場(chǎng)景，我們通常會(huì)選擇并行的方式優(yōu)化，Demo 代碼如下：
public static void main(String[] args) {
// 僅簡(jiǎn)單舉例，在生產(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)有功能可以滿足我們?cè)V求。但當(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í)過長(zhǎng)的 compute(x) 任務(wù)結(jié)果放棄，僅處理在指定時(shí)間內(nèi)完成的任務(wù)，盡可能保證服務(wù)可用。
那么以上代碼的耗時(shí)由耗時(shí)最長(zhǎng)的服務(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) {
// 僅簡(jiǎn)單舉例，在生產(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 請(qǐng)求時(shí)，我們可以設(shè)置一個(gè) JSF 超時(shí)，以達(dá)到異步超時(shí)效果。
當(dāng)請(qǐng)求是一個(gè) R2M 請(qǐng)求時(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í)長(zhǎng)短受 CPU 影響較大。
某些 CPU 使用率高的情況下，就會(huì)出現(xiàn)異步任務(wù)沒能觸發(fā)拋出異常中斷，導(dǎo)致我們無法準(zhǔn)確控制超時(shí)時(shí)間。對(duì)上游來說，本次請(qǐng)求全部失敗。
解決方式
JDK 9
這類問題非常常見，如大促場(chǎng)景，服務(wù)器 CPU 瞬間升高就會(huì)出現(xiàn)以上問題。
那么如何解決呢？其實(shí) JDK 的開發(fā)大佬們?cè)缬醒芯?。?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ù)上述邏輯簡(jiǎn)單實(shí)現(xiàn)一個(gè)工具類來輔助。
以下是我們營(yíng)銷自己的工具類以及用法，貼出來給大家作為參考，大家也可以自己寫的更優(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 異常完成之前等待多長(zhǎng)時(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);
}
}
}總結(jié)
在 JDK 8 場(chǎng)景下，現(xiàn)有超時(shí)中斷的做法依賴于任務(wù)本身的超時(shí)實(shí)現(xiàn)，當(dāng)任務(wù)本身的超時(shí)失效，或者不夠精確時(shí)，并沒有很好的手段來中斷任務(wù)。因此本文給出一種讓 CompletableFuture 支持異步超時(shí)的實(shí)現(xiàn)方案實(shí)現(xiàn)思路，僅供大家參考。
