Java CompletableFuture如何實現(xiàn)超時功能

更新時間：2025年01月28日 09:04:46 作者：樺說編程

這篇文章主要為大家介紹了實現(xiàn)超時功能的基本思路以及CompletableFuture（之后簡稱CF）是如何通過代碼實現(xiàn)超時功能的,需要的小伙伴可以了解下

由于網(wǎng)絡波動或者連接節(jié)點下線等種種問題，對于大多數(shù)網(wǎng)絡異步任務的執(zhí)行通常會進行超時限制，在異步編程中是一個常見的問題。本文主要討論實現(xiàn)超時功能的基本思路以及CompletableFuture（之后簡稱CF）是如何通過代碼實現(xiàn)超時功能的。

基本思路

兩個任務，兩個線程：原有任務，超時任務
原有的任務正常執(zhí)行，寫入正常結果，原有任務執(zhí)行成功取消超時任務
超時時取消原有任務，寫入結果為超時異?；蛘吣J值
競態(tài)條件下保證結果寫入的原子性和只寫一次

CompletableFuture 的實現(xiàn)

1. 基本實現(xiàn)流程

// JDK9新增的超時方法
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;
}

// CF的內(nèi)部類
    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());
        }
    }

分析代碼得知，whenComplete方法添加了正常結束的回調(diào)，取消超時任務。

超時任務通過Delayer.delay創(chuàng)建，超時時執(zhí)行Timeout::run方法，即寫入結果為TimeoutException。

下面來看下Dalayer的具體實現(xiàn)：

/**
 * Singleton delay scheduler, used only for starting and
 * cancelling tasks.
 */
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("CompletableFutureDelayScheduler");
            return t;
        }
    }

    static final ScheduledThreadPoolExecutor delayer;
    static {
        (delayer = new ScheduledThreadPoolExecutor(
            1, new DaemonThreadFactory())).
            setRemoveOnCancelPolicy(true);
    }
}

Delayer是一個單例對象，專門用于執(zhí)行延遲任務，減少了內(nèi)存占用。ScheduledThreadPoolExecutor 的配置為單線程，設置了removeOnCancelPolicy，表示取消延遲任務時，任務從延遲隊列刪除。這里的延遲隊列為默認的執(zhí)行器實現(xiàn)：

public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue(), threadFactory);
}

ScheduledThreadPoolExecutor 底層使用延遲隊列DelayedWorkQueue，延遲隊列底層依賴于索引優(yōu)先隊列，刪除操作的時間復雜度為o(logn)。

下面來看下Canceller的具體實現(xiàn)：

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 (f != null && !f.isDone())
            f.cancel(false);
    }
}

canceller實際上是一個回調(diào)函數(shù)，原有任務完成后觸發(fā)，會取消相關超時任務。

2. 靜態(tài)條件分析

下面是寫入CF的實現(xiàn)代碼片段：

				// 超時結束        
        if (f != null && !f.isDone())
            f.completeExceptionally(new TimeoutException());
        // 取消任務
        if (f != null && !f.isDone())
            f.cancel(false);
				// CF 原有任務的寫入不由orTimeout方法控制，以下為一個示例
						Thread.sleep(1000);
						f.complete(u);

對于CF的檢查實際上不能保證原子性，因為這種檢查-再計算的模式需要同步塊的保護，而CF底層并沒有這種實現(xiàn)。所以，if語句檢查任務未完成，之后執(zhí)行代碼時，任務可能已經(jīng)完成了。不過這種檢查也有一定的好處，因為CF保證了結果寫入后，isDone方法必然為true，從而避免執(zhí)行不必要的代碼。

completeExceptionally 方法和 complete 方法可能同時執(zhí)行，CF 通過CAS操作保證了結果寫入的原子性。

// 異常結果實現(xiàn)
final boolean internalComplete(Object r) { // CAS from null to r
    return RESULT.compareAndSet(this, null, r);
}
// 正常結果實現(xiàn)
final boolean completeValue(T t) {
    return RESULT.compareAndSet(this, null, (t == null) ? NIL : t);
}

public boolean isDone() {
    return result != null;
}

3. 內(nèi)存泄露 bug

在 JDK21之前的CF實現(xiàn)中，存在內(nèi)存泄露的bug，具體描述詳見 https://bugs.openjdk.org/browse/JDK-8303742 ，目前筆者僅在JDK21 中發(fā)現(xiàn)代碼已修復（不考慮非LTS版本）。作為bug，后續(xù)發(fā)布的 JDK 子版本可能會修復這個問題。

這個bug在如下代碼中：

// 取消任務，JDK21之前的實現(xiàn)會檢查異常結果
if (ex == null && f != null && !f.isDone())
    f.cancel(false);

當正常任務異常結束時，不會取消延遲隊列中的任務，最終會導致內(nèi)存泄露。若項目中存在多個長時間超時CF任務，內(nèi)存泄露的情況會更明顯。

public class LeakDemo {
    public static void main(String[] args) {
        while (true) {
            new CompletableFuture<>().orTimeout(1, TimeUnit.HOURS).completeExceptionally(new Exception());
        }
    }
}

執(zhí)行以上代碼會報OOM錯誤，你可以在自己的編程環(huán)境中進行測試。