欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Java CompletableFuture 異步超時實現(xiàn)深入研究

 更新時間:2023年02月08日 15:10:26   作者:京東云開發(fā)者  
這篇文章主要為大家介紹了Java CompletableFuture 異步超時實現(xiàn)深入研究,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪

前言

作者:京東科技 張?zhí)熨n

JDK 8 是一次重大的版本升級,新增了非常多的特性,其中之一便是 CompletableFuture。自此從 JDK 層面真正意義上的支持了基于事件的異步編程范式,彌補了 Future 的缺陷。

在我們的日常優(yōu)化中,最常用手段便是多線程并行執(zhí)行。這時候就會涉及到 CompletableFuture 的使用。

常見使用方式

下面舉例一個常見場景。

假如我們有兩個 RPC 遠程調(diào)用服務,我們需要獲取兩個 RPC 的結果后,再進行后續(xù)邏輯處理。

public static void main(String[] args) {
    // 任務 A,耗時 2 秒
    int resultA = compute(1);
    // 任務 B,耗時 2 秒
    int resultB = compute(2);
    // 后續(xù)業(yè)務邏輯處理
    System.out.println(resultA + resultB);
}

可以預估到,串行執(zhí)行最少耗時 4 秒,并且 B 任務并不依賴 A 任務結果。

對于這種場景,我們通常會選擇并行的方式優(yōu)化,Demo 代碼如下:

public static void main(String[] args) {
    // 僅簡單舉例,在生產(chǎn)代碼中可別這么寫!
    // 統(tǒng)計耗時的函數(shù)
    time(() -> {
        CompletableFuture<Integer> result = Stream.of(1, 2)
                                                  // 創(chuàng)建異步任務
                                                  .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("任務執(zhí)行異常");
        }
    });
}
輸出:
[async-1]: 任務執(zhí)行開始:1
[async-2]: 任務執(zhí)行開始:2
[async-1]: 任務執(zhí)行完成:1
[async-2]: 任務執(zhí)行完成:2
結果:3
耗時:2 秒

可以看到耗時變成了 2 秒。

存在的問題

分析

看上去 CompletableFuture 現(xiàn)有功能可以滿足我們訴求。但當我們引入一些現(xiàn)實常見情況時,一些潛在的不足便暴露出來了。

compute(x) 如果是一個根據(jù)入?yún)⒉樵冇脩裟愁愋蛢?yōu)惠券列表的任務,我們需要查詢兩種優(yōu)惠券并組合在一起返回給上游。假如上游要求我們 2 秒內(nèi)處理完畢并返回結果,但 compute(x) 耗時卻在 0.5 秒 ~ 無窮大波動。這時候我們就需要把耗時過長的 compute(x) 任務結果放棄,僅處理在指定時間內(nèi)完成的任務,盡可能保證服務可用。

那么以上代碼的耗時由耗時最長的服務決定,無法滿足現(xiàn)有訴求。通常我們會使用 get(long timeout, TimeUnit unit) 來指定獲取結果的超時時間,并且我們會給 compute(x) 設置一個超時時間,達到后自動拋異常來中斷任務。

public static void main(String[] args) {
    // 僅簡單舉例,在生產(chǎn)代碼中可別這么寫!
    // 統(tǒng)計耗時的函數(shù)
    time(() -> {
        List<CompletableFuture<Integer>> result = Stream.of(1, 2)
                                                        // 創(chuàng)建異步任務,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("任務執(zhí)行異?;虺瑫r");
            }
        }
        System.out.println("結果:" + res);
    });
}
輸出:
[async-2]: 任務執(zhí)行開始:2
[async-1]: 任務執(zhí)行開始:1
[async-1]: 任務執(zhí)行完成:1
任務執(zhí)行異?;虺瑫r
結果:1
耗時:2 秒

可以看到,只要我們能夠給 compute(x) 設置一個超時時間將任務中斷,結合 get、getNow 等獲取結果的方式,就可以很好地管理整體耗時。

那么問題也就轉變成了,如何給任務設置異步超時時間呢?

現(xiàn)有做法

當異步任務是一個 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)) {
        // 等待時間內(nèi)任務完成,反序列化獲取結果
        return this.getNow();
    }
    this.setDoneTime();
    // 超時拋出異常
    throw this.clientTimeoutException(false);
}

當這個任務剛好卡在超時邊緣完成時,這個任務的耗時時間就變成了超時時間 + 獲取結果時間。而獲取結果(反序列化)作為純本地計算操作,耗時長短受 CPU 影響較大。

某些 CPU 使用率高的情況下,就會出現(xiàn)異步任務沒能觸發(fā)拋出異常中斷,導致我們無法準確控制超時時間。對上游來說,本次請求全部失敗。

解決方式

JDK 9

這類問題非常常見,如大促場景,服務器 CPU 瞬間升高就會出現(xiàn)以上問題。

那么如何解決呢?其實 JDK 的開發(fā)大佬們早有研究。在 JDK 9,CompletableFuture 正式提供了 orTimeout、completeTimeout 方法,來準確實現(xiàn)異步超時控制。

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 其實現(xiàn)原理是通過一個定時任務,在給定時間之后拋出異常。如果任務在指定時間內(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())
            // 拋出超時異常
            f.completeExceptionally(new TimeoutException());
    }
}

通過源碼可以看到,Timeout 是一個實現(xiàn) Runnable 的類,run() 方法負責給傳入的異步任務通過 completeExceptionally CAS 賦值異常,將任務標記為異常完成。

那么誰來觸發(fā)這個 run() 方法呢?我們看下 Delayer 的實現(xiàn)。

static final class Delayer {
    static ScheduledFuture<?> delay(Runnable command, long delay,
                                    TimeUnit unit) {
        // 到時間觸發(fā) 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 其實就是一個單例定時調(diào)度器,Delayer.delay(new Timeout(this), timeout, unit) 通過 ScheduledThreadPoolExecutor 實現(xiàn)指定時間后觸發(fā) Timeout 的 run() 方法。

到這里就已經(jīng)實現(xiàn)了超時拋出異常的操作。但當任務完成時,就沒必要觸發(fā) Timeout 了。因此我們還需要實現(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 (ex == null && f != null && !f.isDone())
        // 3 未觸發(fā)拋異常任務則取消
            f.cancel(false);
    }
}

當任務執(zhí)行完成,或者任務執(zhí)行異常時,我們也就沒必要拋出超時異常了。因此我們可以把 delayer.schedule(command, delay, unit) 返回的定時超時任務取消,不再觸發(fā) Timeout。 當我們的異步任務完成,并且定時超時任務未完成的時候,就是我們?nèi)∠臅r機。因此我們可以通過 whenComplete(BiConsumer<? super T, ? super Throwable> action) 來完成。

Canceller 就是一個 BiConsumer 的實現(xiàn)。其持有了 delayer.schedule(command, delay, unit) 返回的定時超時任務,accept(Object ignore, Throwable ex) 實現(xiàn)了定時超時任務未完成后,執(zhí)行 cancel(boolean mayInterruptIfRunning) 取消任務的操作。

JDK 8

如果我們使用的是 JDK 9 或以上,我們可以直接用 JDK 的實現(xiàn)來完成異步超時操作。那么 JDK 8 怎么辦呢?

其實我們也可以根據(jù)上述邏輯簡單實現(xiàn)一個工具類來輔助。

以下是我們營銷自己的工具類以及用法,貼出來給大家作為參考,大家也可以自己寫的更優(yōu)雅一些~

調(diào)用方式:

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 在出現(xiàn) TimeoutException 異常完成之前等待多長時間,以 {@code unit} 為單位
     * @param unit    一個 {@link TimeUnit},結合 {@code timeout} 參數(shù),表示給定粒度單位的持續(xù)時間
     * @return 入?yún)⒌?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);
            }
        }
    }
    /**
     * 單例延遲調(diào)度器,僅用于啟動和取消任務,一個線程就足夠
     */
    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 異步超時實現(xiàn)深入研究的詳細內(nèi)容,更多關于Java CompletableFuture 異步超時的資料請關注腳本之家其它相關文章!

相關文章

  • Spring整合CXF webservice restful實例詳解

    Spring整合CXF webservice restful實例詳解

    這篇文章主要為大家詳細介紹了Spring整合CXF webservice restful的實例,具有一定的參考價值,感興趣的小伙伴們可以參考一下
    2017-08-08
  • Spring異常實現(xiàn)統(tǒng)一處理的方法

    Spring異常實現(xiàn)統(tǒng)一處理的方法

    這篇文章主要介紹了Spring異常實現(xiàn)統(tǒng)一處理的方法,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習吧
    2022-12-12
  • Spring?中?PageHelper?不生效問題及解決方法

    Spring?中?PageHelper?不生效問題及解決方法

    這篇文章主要介紹了Spring?中?PageHelper?不生效問題,使用這個插件時要注意版本的問題,不同的版本可能 PageHelper 不會生效,本文結合示例代碼給大家介紹的非常詳細,需要的朋友可以參考下
    2022-12-12
  • springmvc如何使用POJO作為參數(shù)

    springmvc如何使用POJO作為參數(shù)

    這篇文章主要介紹了springmvc如何使用POJO作為參數(shù),文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下
    2020-01-01
  • Java RabbitMQ的三種Exchange模式

    Java RabbitMQ的三種Exchange模式

    這篇文章主要介紹了Java RabbitMQ的三種Exchange模式,分別為Direct模式、Fanout模式、Topic模式,Rabbit的Direct Exchange模式是指消息發(fā)送導RouteKey中指定的Queue,Direct模式可以使用Rabbit自帶的Exchange
    2022-08-08
  • 使用springboot單元測試對weblistener的加載測試

    使用springboot單元測試對weblistener的加載測試

    這篇文章主要介紹了使用springboot單元測試對weblistener的加載測試,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2021-10-10
  • springboot2.0+elasticsearch5.5+rabbitmq搭建搜索服務的坑

    springboot2.0+elasticsearch5.5+rabbitmq搭建搜索服務的坑

    這篇文章主要介紹了springboot2.0+elasticsearch5.5+rabbitmq搭建搜索服務的坑,小編覺得挺不錯的,現(xiàn)在分享給大家,也給大家做個參考。一起跟隨小編過來看看吧
    2018-06-06
  • Spring MVC 與 CORS跨域的詳細介紹

    Spring MVC 與 CORS跨域的詳細介紹

    本文介紹了 CORS 的知識以及如何在 Spring MVC 中配置 CORS,具有一定的參考價值,感興趣的小伙伴們可以參考一下
    2017-05-05
  • Spring Cloud Gateway 使用JWT工具類做用戶登錄校驗功能

    Spring Cloud Gateway 使用JWT工具類做用戶登錄校驗功能

    這篇文章主要介紹了Spring Cloud Gateway 使用JWT工具類做用戶登錄校驗的示例代碼,本文給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下
    2021-01-01
  • SpringBoot后端上傳文件類型檢測方式

    SpringBoot后端上傳文件類型檢測方式

    這篇文章主要介紹了SpringBoot后端上傳文件類型檢測方式,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2022-03-03

最新評論