本文主要是介绍Java CompletableFuture如何实现超时功能，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
《JavaCompletableFuture如何实现超时功能》：本文主要介绍实现超时功能的基本思路以及CompletableFuture（之后简称CF）是如何通过代码实现超时功能的,需要的...
由于网络波动或者连接节点下线等种种问题，对于大多数网络异步任务的执行通常会进行超时限制，在异步编程中是一个常见的问题。本文主要讨论实现超时功能的基本思路以及CompletableFuture（之后简称CF）是如何通过代码实现超时功能的。
基本思路
- 两个任务，两个线程：原有任务，超时任务
- 原有的任务正常执行，写入正常结果，原有任务执行成功取消超时任务
- 超时时取消原有任务，写入结果为超时异常或者默认值
- 竞态条件下保证结果写入的原子性和只写一次
CompletableFuture 的实现
1. 基本实现流程
// 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的内部类
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方法添加了正常结束的回调，取消超时任务。
超时任务通过Delayer.delay创建，超时时执行Timeout::run方法，即写入结果为TimeoutException。
下面来看下Dalayer的具体实现：
/** * 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是一个单例对象，专门用于执行延迟任务，减少了内存占用。ScheduledThreadPoolExecutor 的配置为单线程，设置了removeOnCancelPolicy，表示取消延迟任务时，任务从延迟队列删除。这里的延迟队列为默认的执行器实现：
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE,
DEFAULT_KEEPALIVE_MILLIS, MILLISECONDSjs,
new DelayedworkQueue(), threadFactory);
}
ScheduledThreadPoolExecutor 底层使用延迟队列DelayedWorkQueue，延迟队列底层依赖于索引优先队列，删除操作的时间复杂度为o(logn)。
下面来看下Canceller的具体实现：
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实际上是一个回调函数，原有任务完成后触发，会取消相关超时任务。
2. 静态条件分析
下面是写入CF的实现代码片段：
// 超时结束
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底层并没有这种实现。所以，if语句检查任务未完成，之后执行代码时，任务可能已经完成了。不过这种检查也有一定的好处，因为CF保证了结果写入后，isDone方法必然为true，从而避免执行不必要的代码。
completeExceptionally 方法和 complete 方法可能同时执行，CF 通过CAS操作保证了结果写入的原子性。
// 异常结果实现
final boolean internalComplete(Object r) { // CAS from null to r
return RESULT.compareAndSet(this, null, r);
}
// 正常结果实现
final boolean completeValue(T t) {
return RESULT.compareAndSet(this, null, (t == null) ? NIL : t);
}
public boolean isDone() {
return result != null;
}
3. 内存泄露 bug
在 JDK21之前的CF实现中，存在内存泄露的bug，具体描述详见 https://bugs.openjdk.org/browse/JDK-8303742 ，目前笔者仅在JDK21 中发现代码已修复（不考虑非LTS版本）。作为bug，后续发布的 JDK 子版本可能会修复这个问题。
这个bug在如下代码中：
// 取消任务，JDK21之前的实现会检查异常结果 if (ex == null && f != null && !f.isDone()) China编程 f.cancel(false);
当正常任务异常结束时，不会取消延迟队列中的任务，最终会导致内存泄露。若项目中存在多个长时间超时CF任务，内存泄露的情况会更明显。
public class LeakDemo {
public static void main(String[] args) {
while (true) {
new CompletabphpleFuture<>().orTimeout(1, TimeUnit.HOURS).completeExceptionally(new Exception());
}
}
}
执行以上代码会报OOM错误，你可以在自己的编程环境中进行测试。
4. JDK8如何实现超时任务
JDK8中CompletableFuture并不支持超时任务，笔者推荐使用CFFU类库，其是CF的增强类库，支持在JDK8环境中使用高版本的功能。另一种方案使用 Guava 提供的 ListenableFuture。当然你也可以参照JDK21中编程的代码自己实现。
到此这篇关于Java CompletableFuture如何实现超时功能的文章就介绍到这了,更多相关Java CompletableFuture超时内容请搜索China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持China编程(www.chinasem.cn)！
这篇关于Java CompletableFuture如何实现超时功能的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
