Java CompletableFuture 非同步超時實現探索

来源:https://www.cnblogs.com/Jcloud/archive/2023/02/08/17100990.html
-Advertisement-
Play Games

JDK 8 是一次重大的版本升級，新增了非常多的特性，其中之一便是 CompletableFuture。自此從 JDK 層面真正意義上的支持了基於事件的非同步編程範式，彌補了 Future 的缺陷。 在我們的日常優化中，最常用手段便是多線程並行執行。這時候就會涉及到 CompletableFutur... ...


作者：京東科技 張天賜

前言

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);
        }
    }
}

參考資料

  1. JEP 266: JDK 9 併發包更新提案

您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • JavaScript 高階函數
    JavaScript 中的高階函數是一種接受函數作為輸入或返回函數作為輸出的函數。高階函數可以用於創建抽象概念，例如柯里化、組合和管道。 ...
  • 如何創建乾凈的C++工程
    如何創建有層次的C++工程目錄 大多數人創建C++工程採用預設設置，這樣會導致工程內文件散亂，毫無可觀性。那麼如何去創建一個可觀性搶的工程目錄呢？以下將為大家解惑。 1、快速創建預設工程並簡單處理 a、新手因不理解具體的項目工程有哪些文件目錄，所以我建議直接創建一個預設的項目，然後關閉當前目錄創建這 ...
  • 讀Java實戰（第二版）筆記05_Collection API的增強功能
    一、應用介紹 ”所有網頁都是客戶端“為fusion app的核心概念，讓我們利用它可以快速的做出屬於自己的應用。由於作者已經停止更新原版app，文章採用FA 重製版完成。 成品展示： 體驗下載： 下載鏈接:https://wwp.lanzoup.com/iIKXF0mxu1gd 提取碼 : xcvb ...
  • 《Terraform 101 從入門到實踐》 前言
    《Terraform 101 從入門到實踐》這本小冊在南瓜慢說官方網站和GitHub兩個地方同步更新，書中的示例代碼也是放在GitHub上，方便大家參考查看。 Terraform 101 從入門到實踐 Terraform作為基礎設施即代碼（Infrastructure as Code，很簡稱IaC） ...
  • Python教程：selenium模塊用法教程
    1.介紹 selenium最初是一個自動化測試工具,而爬蟲中使用它主要是為瞭解決requests無法直接執行JavaScript代碼的問題 selenium本質是通過驅動瀏覽器，完全模擬瀏覽器的操作，比如跳轉、輸入、點擊、下拉等，來拿到網頁渲染之後的結果，可支持多種瀏覽器 from selenium ...
  • if條件判斷和for迴圈
    條件判斷if 簡單的if判斷 package main import "fmt" func main(){ age := 18 if age >=18 { fmt.Println("呦呵長大了") }else { fmt.Println("還沒長大") } } 多分支if package main ...
  • 分散式配置nacos搭建踩坑指南(上)
    春節電影聽巳月說都還可以，我不信，我覺得還是要看看看過的觀眾怎麼說，於是我點開了流浪地球2 … 看起來好像不錯的樣子，8.2的評分，三十多億的票房 就是這評價也太多了，那我們今天就把網友對它的評論獲取下來，做成可視化詞雲圖看看大家討論最多的是什麼。 準備工作 使用的環境 Python 3.8 解釋器 ...
  • 《分散式技術原理與演算法解析》學習筆記Day05
    這篇文章主要討論分散式共識，包括什麼是分散式共識以及常用的三種分散式共識演算法：PoW（工作量證明）、PoS（權益證明）和DPoS（委托權益證明）。 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類