Java CompletableFuture如何實(shí)現(xiàn)超時功能
由于網(wǎng)絡(luò)波動或者連接節(jié)點(diǎn)下線等種種問題，對于大多數(shù)網(wǎng)絡(luò)異步任務(wù)的執(zhí)行通常會進(jìn)行超時限制，在異步編程中是一個常見的問題。本文主要討論實(shí)現(xiàn)超時功能的基本思路以及CompletableFuture（之后簡稱CF）是如何通過代碼實(shí)現(xiàn)超時功能的。
基本思路
- 兩個任務(wù)，兩個線程：原有任務(wù)，超時任務(wù)
- 原有的任務(wù)正常執(zhí)行，寫入正常結(jié)果，原有任務(wù)執(zhí)行成功取消超時任務(wù)
- 超時時取消原有任務(wù)，寫入結(jié)果為超時異常或者默認(rèn)值
- 競態(tài)條件下保證結(jié)果寫入的原子性和只寫一次
CompletableFuture 的實(shí)現(xiàn)
1. 基本實(shí)現(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方法添加了正常結(jié)束的回調(diào)，取消超時任務(wù)。
超時任務(wù)通過Delayer.delay創(chuàng)建，超時時執(zhí)行Timeout::run方法，即寫入結(jié)果為TimeoutException。
下面來看下Dalayer的具體實(shí)現(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);
// 守護(hù)線程，當(dāng)主線程關(guān)閉時，自身也關(guān)閉
t.setDaemon(true);
t.setName("CompletableFutureDelayScheduler");
return t;
}
}
static final ScheduledThreadPoolExecutor delayer;
static {
(delayer = new ScheduledThreadPoolExecutor(
1, new DaemonThreadFactory())).
setRemoveOnCancelPolicy(true);
}
}
Delayer是一個單例對象，專門用于執(zhí)行延遲任務(wù)，減少了內(nèi)存占用。ScheduledThreadPoolExecutor 的配置為單線程，設(shè)置了removeOnCancelPolicy，表示取消延遲任務(wù)時，任務(wù)從延遲隊列刪除。這里的延遲隊列為默認(rèn)的執(zhí)行器實(shí)現(xiàn)：
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE,
DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
new DelayedWorkQueue(), threadFactory);
}
ScheduledThreadPoolExecutor 底層使用延遲隊列DelayedWorkQueue，延遲隊列底層依賴于索引優(yōu)先隊列，刪除操作的時間復(fù)雜度為o(logn)。
下面來看下Canceller的具體實(shí)現(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實(shí)際上是一個回調(diào)函數(shù)，原有任務(wù)完成后觸發(fā)，會取消相關(guān)超時任務(wù)。
2. 靜態(tài)條件分析
下面是寫入CF的實(shí)現(xiàn)代碼片段：
// 超時結(jié)束
if (f != null && !f.isDone())
f.completeExceptionally(new TimeoutException());
// 取消任務(wù)
if (f != null && !f.isDone())
f.cancel(false);
// CF 原有任務(wù)的寫入不由orTimeout方法控制，以下為一個示例
Thread.sleep(1000);
f.complete(u);
對于CF的檢查實(shí)際上不能保證原子性，因?yàn)檫@種檢查-再計算的模式需要同步塊的保護(hù)，而CF底層并沒有這種實(shí)現(xiàn)。所以，if語句檢查任務(wù)未完成，之后執(zhí)行代碼時，任務(wù)可能已經(jīng)完成了。不過這種檢查也有一定的好處，因?yàn)镃F保證了結(jié)果寫入后，isDone方法必然為true，從而避免執(zhí)行不必要的代碼。
completeExceptionally 方法和 complete 方法可能同時執(zhí)行，CF 通過CAS操作保證了結(jié)果寫入的原子性。
// 異常結(jié)果實(shí)現(xiàn)
final boolean internalComplete(Object r) { // CAS from null to r
return RESULT.compareAndSet(this, null, r);
}
// 正常結(jié)果實(shí)現(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實(shí)現(xiàn)中，存在內(nèi)存泄露的bug，具體描述詳見 https://bugs.openjdk.org/browse/JDK-8303742 ，目前筆者僅在JDK21 中發(fā)現(xiàn)代碼已修復(fù)（不考慮非LTS版本）。作為bug，后續(xù)發(fā)布的 JDK 子版本可能會修復(fù)這個問題。
這個bug在如下代碼中：
// 取消任務(wù)，JDK21之前的實(shí)現(xiàn)會檢查異常結(jié)果
if (ex == null && f != null && !f.isDone())
f.cancel(false);
當(dāng)正常任務(wù)異常結(jié)束時，不會取消延遲隊列中的任務(wù)，最終會導(dǎo)致內(nèi)存泄露。若項目中存在多個長時間超時CF任務(wù)，內(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)境中進(jìn)行測試。
4. JDK8如何實(shí)現(xiàn)超時任務(wù)
JDK8中CompletableFuture并不支持超時任務(wù)，筆者推薦使用CFFU類庫，其是CF的增強(qiáng)類庫，支持在JDK8環(huán)境中使用高版本的功能。另一種方案使用 Guava 提供的 ListenableFuture。當(dāng)然你也可以參照J(rèn)DK21中的代碼自己實(shí)現(xiàn)。
到此這篇關(guān)于Java CompletableFuture如何實(shí)現(xiàn)超時功能的文章就介紹到這了,更多相關(guān)Java CompletableFuture超時內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要為大家詳細(xì)介紹了SpringMVC上傳文件FileUpload的使用方法，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-11-11
SpringBoot實(shí)現(xiàn)登錄攔截的示例代碼
如果我們不進(jìn)行登錄攔截的話,即使我們跳過登錄頁面直接去訪問任意一個頁面也能訪問成功,那么登錄功能就沒有意義,同時也會存在安全問題,本文就來介紹一下SpringBoot登錄攔截,具有一定的參考價值,感興趣的可以了解一下2023-09-09
這篇文章主要介紹了java將word轉(zhuǎn)pdf的相關(guān)資料,文中講解了使用Aspose-Words工具將Word文檔轉(zhuǎn)換為PDF的優(yōu)劣,并提供了一種在Java項目中使用Aspose-Words進(jìn)行Word轉(zhuǎn)PDF的示例方法,需要的朋友可以參考下2025-01-01
JDK8中String的intern()方法實(shí)例詳細(xì)解讀
String字符串在我們?nèi)粘ｉ_發(fā)中最常用的，當(dāng)然還有他的兩個兄弟StringBuilder和StringBuilder，接下來通過本文給大家介紹JDK8中String的intern()方法詳細(xì)解讀,需要的朋友可以參考下2022-09-09
Spring實(shí)戰(zhàn)之Bean的作用域request用法分析
這篇文章主要介紹了Spring實(shí)戰(zhàn)之Bean的作用域request用法,結(jié)合實(shí)例形式分析了spring中Bean的request作用域相關(guān)使用技巧與操作注意事項,需要的朋友可以參考下2019-11-11
這篇文章主要介紹了新版POI獲取日期類型cell值過程圖解,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-10-10
本文給大家分享的是java實(shí)現(xiàn)讀取、刪除文件夾下的文件，其中File.delete()用于刪除“某個文件或者空目錄”！所以要刪除某個目錄及其中的所有文件和子目錄，要進(jìn)行遞歸刪除，有需要的小伙伴可以參考下。2015-05-05
在本文中，將從大到小，從外到內(nèi)，總結(jié)Java編程中的命名規(guī)范。文中將會涉及到日常工作中常見的命名示例，如包命名，類命名，接口命名，方法命名，變量命名，常類命名，抽象類命名，異常類命名以及擴(kuò)展類命名等。2021-09-09
這篇文章主要介紹了舉例說明Java中代碼塊的執(zhí)行順序,包括靜態(tài)屬性和非靜態(tài)屬性以及構(gòu)造函數(shù)等相關(guān)的執(zhí)行先后,需要的朋友可以參考下2015-07-07
SpringBoot中HttpSessionListener的簡單使用方式
這篇文章主要介紹了SpringBoot中HttpSessionListener的簡單使用方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-03-03