作者：京東科技 張天賜
前言
JDK 8 是一次重大的版本升級，新增了非常多的特性，其中之一便是 CompletableFuture，自此從 JDK 層面真正意義上的支持了基于事件的異步編程范式，彌補了 Future 的缺陷，
在我們的日常優化中，最常用手段便是多執行緒并行執行，這時候就會涉及到 CompletableFuture 的使用，
常見使用方式
下面舉例一個常見場景，
假如我們有兩個 RPC 遠程呼叫服務，我們需要獲取兩個 RPC 的結果后，再進行后續邏輯處理，
public static void main(String[] args) {
// 任務 A，耗時 2 秒
int resultA = compute(1);
// 任務 B，耗時 2 秒
int resultB = compute(2);
// 后續業務邏輯處理
System.out.println(resultA + resultB);
}
可以預估到，串行執行最少耗時 4 秒，并且 B 任務并不依賴 A 任務結果，
對于這種場景，我們通常會選擇并行的方式優化，Demo 代碼如下：
public static void main(String[] args) {
// 僅簡單舉例，在生產代碼中可別這么寫！
// 統計耗時的函式
time(() -> {
CompletableFuture<Integer> result = Stream.of(1, 2)
// 創建異步任務
.map(x -> CompletableFuture.supplyAsync(() -> compute(x), executor))
// 聚合
.reduce(CompletableFuture.completedFuture(0), (x, y) -> x.thenCombineAsync(y, Integer::sum, executor));
// 等待結果
try {
System.out.println("結果：" + result.get());
} catch (ExecutionException | InterruptedException e) {
System.err.println("任務執行例外");
}
});
}
輸出：
[async-1]: 任務執行開始：1
[async-2]: 任務執行開始：2
[async-1]: 任務執行完成：1
[async-2]: 任務執行完成：2
結果：3
耗時：2 秒
可以看到耗時變成了 2 秒，
存在的問題
分析
看上去 CompletableFuture 現有功能可以滿足我們訴求，但當我們引入一些現實常見情況時，一些潛在的不足便暴露出來了，
compute(x) 如果是一個根據入參查詢用戶某型別優惠券串列的任務，我們需要查詢兩種優惠券并組合在一起回傳給上游，假如上游要求我們 2 秒內處理完畢并回傳結果，但 compute(x) 耗時卻在 0.5 秒 ~ 無窮大波動，這時候我們就需要把耗時過長的 compute(x) 任務結果放棄，僅處理在指定時間內完成的任務，盡可能保證服務可用，
那么以上代碼的耗時由耗時最長的服務決定，無法滿足現有訴求，通常我們會使用 get(long timeout, TimeUnit unit) 來指定獲取結果的超時時間，并且我們會給 compute(x) 設定一個超時時間，達到后自動拋例外來中斷任務，
public static void main(String[] args) {
// 僅簡單舉例，在生產代碼中可別這么寫！
// 統計耗時的函式
time(() -> {
List<CompletableFuture<Integer>> result = Stream.of(1, 2)
// 創建異步任務，compute(x) 超時拋出例外
.map(x -> CompletableFuture.supplyAsync(() -> compute(x), executor))
.toList();
// 等待結果
int res = 0;
for (CompletableFuture<Integer> future : result) {
try {
res += future.get(2, SECONDS);
} catch (ExecutionException | InterruptedException | TimeoutException e) {
System.err.println("任務執行例外或超時");
}
}
System.out.println("結果：" + res);
});
}
輸出：
[async-2]: 任務執行開始：2
[async-1]: 任務執行開始：1
[async-1]: 任務執行完成：1
任務執行例外或超時
結果：1
耗時：2 秒
可以看到，只要我們能夠給 compute(x) 設定一個超時時間將任務中斷，結合 get、getNow 等獲取結果的方式，就可以很好地管理整體耗時，
那么問題也就轉變成了，如何給任務設定異步超時時間呢？
現有做法
當異步任務是一個 RPC 請求時，我們可以設定一個 JSF 超時，以達到異步超時效果，
當請求是一個 R2M 請求時，我們也可以控制 R2M 連接的最大超時時間來達到效果，
這么看好像我們都是在依賴三方中間件的能力來管理任務超時時間？那么就存在一個問題，中間件超時控制能力有限，如果異步任務是中間件 IO 操作 + 本地計算操作怎么辦？
用 JSF 超時舉一個具體的例子，反編譯 JSF 的獲取結果代碼如下：
public V get(long timeout, TimeUnit unit) throws InterruptedException {
// 配置的超時時間
timeout = unit.toMillis(timeout);
// 剩余等待時間
long remaintime = timeout - (this.sentTime - this.genTime);
if (remaintime <= 0L) {
if (this.isDone()) {
// 反序列化獲取結果
return this.getNow();
}
} else if (this.await(remaintime, TimeUnit.MILLISECONDS)) {
// 等待時間內任務完成，反序列化獲取結果
return this.getNow();
}
this.setDoneTime();
// 超時拋出例外
throw this.clientTimeoutException(false);
}
當這個任務剛好卡在超時邊緣完成時，這個任務的耗時時間就變成了超時時間 + 獲取結果時間，而獲取結果（反序列化）作為純本地計算操作，耗時長短受 CPU 影響較大，
某些 CPU 使用率高的情況下，就會出現異步任務沒能觸發拋出例外中斷，導致我們無法準確控制超時時間，對上游來說，本次請求全部失敗，
解決方式
JDK 9
這類問題非常常見，如大促場景，服務器 CPU 瞬間升高就會出現以上問題，
那么如何解決呢？其實 JDK 的開發大佬們早有研究，在 JDK 9，CompletableFuture 正式提供了 orTimeout、completeTimeout 方法，來準確實作異步超時控制，
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 其實作原理是通過一個定時任務，在給定時間之后拋出例外，如果任務在指定時間內完成，則取消拋例外的操作，
以上代碼我們按執行順序來看下：
首先執行 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())
// 拋出超時例外
f.completeExceptionally(new TimeoutException());
}
}
通過原始碼可以看到，Timeout 是一個實作 Runnable 的類，run() 方法負責給傳入的異步任務通過 completeExceptionally CAS 賦值例外，將任務標記為例外完成，
那么誰來觸發這個 run() 方法呢？我們看下 Delayer 的實作，
static final class Delayer {
static ScheduledFuture<?> delay(Runnable command, long delay,
TimeUnit unit) {
// 到時間觸發 command 任務
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 其實就是一個單例定時調度器，Delayer.delay(new Timeout(this), timeout, unit) 通過 ScheduledThreadPoolExecutor 實作指定時間后觸發 Timeout 的 run() 方法，
到這里就已經實作了超時拋出例外的操作，但當任務完成時，就沒必要觸發 Timeout 了，因此我們還需要實作一個取消邏輯，
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.cancel(false);
}
}
當任務執行完成，或者任務執行例外時，我們也就沒必要拋出超時例外了，因此我們可以把 delayer.schedule(command, delay, unit) 回傳的定時超時任務取消，不再觸發 Timeout， 當我們的異步任務完成，并且定時超時任務未完成的時候，就是我們取消的時機，因此我們可以通過 whenComplete(BiConsumer<? super T, ? super Throwable> action) 來完成，
Canceller 就是一個 BiConsumer 的實作，其持有了 delayer.schedule(command, delay, unit) 回傳的定時超時任務，accept(Object ignore, Throwable ex) 實作了定時超時任務未完成后，執行 cancel(boolean mayInterruptIfRunning) 取消任務的操作，
JDK 8
如果我們使用的是 JDK 9 或以上，我們可以直接用 JDK 的實作來完成異步超時操作，那么 JDK 8 怎么辦呢？
其實我們也可以根據上述邏輯簡單實作一個工具類來輔助，
以下是我們營銷自己的工具類以及用法，貼出來給大家作為參考，大家也可以自己寫的更優雅一些~
呼叫方式：
CompletableFutureExpandUtils.orTimeout(異步任務, 超時時間, 時間單位);
工具類原始碼：
package com.jd.jr.market.reduction.util;
import com.jdpay.market.common.exception.UncheckedException;
import java.util.concurrent.*;
import java.util.function.BiConsumer;
/**
* CompletableFuture 擴展工具
*
* @author zhangtianci7
*/
public class CompletableFutureExpandUtils {
/**
* 如果在給定超時之前未完成，則例外完成此 CompletableFuture 并拋出 {@link TimeoutException} ，
*
* @param timeout 在出現 TimeoutException 例外完成之前等待多長時間，以 {@code unit} 為單位
* @param unit 一個 {@link TimeUnit}，結合 {@code timeout} 引數，表示給定粒度單位的持續時間
* @return 入參的 CompletableFuture
*/
public static <T> CompletableFuture<T> orTimeout(CompletableFuture<T> future, long timeout, TimeUnit unit) {
if (null == unit) {
throw new UncheckedException("時間的給定粒度不能為空");
}
if (null == future) {
throw new UncheckedException("異步任務不能為空");
}
if (future.isDone()) {
return future;
}
return future.whenComplete(new Canceller(Delayer.delay(new Timeout(future), timeout, unit)));
}
/**
* 超時時例外完成的操作
*/
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());
}
}
}
/**
* 取消不需要的超時的操作
*/
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);
}
}
}
/**
* 單例延遲調度器，僅用于啟動和取消任務，一個執行緒就足夠
*/
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 并發包更新提案
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/543270.html
標籤：其他
上一篇：SpringBoot