CompletableFuture 超时功能有大坑！使用不当直接生产事故！

CompletableFuture 超时功能有大坑！使用不当直接生产事故！

本文未经允许禁止转载！

上一篇文章《如何实现超时功能(以CompletableFuture为例)》中我们讨论了 CompletableFuture 超时功能的具体实现，从整体实现来说，JDK21前的版本有着内存泄露的bug，不过很少对实际生产有影响，因为任务的编排涉及的对象并不多，少量内存泄露最终会被回收掉。从单一功能内聚的角度来说，超时功能的实现是没有问题；然而由于并发编程的复杂性，可能会出现 Delayer 线程延迟执行的情况。本文将详细复现与讨论 CompletableFuture 超时功能的大坑，同时提供一些最佳实践指导。

2024年9月8日更新：CFFU 开源项目负责人李鼎(Jerry Lee) 更新了代码示例，点击这里查看。

1. 问题复现

感谢 CFFU 开源项目负责人李鼎(Jerry Lee) 提供代码：

public class CfDelayDysfunctionDemo {public static void main(String[] args) {dysfunctionDemo();System.out.println();cffuOrTimeoutFixDysfunctionDemo();}private static void dysfunctionDemo() {logWithTimeAndThread("dysfunctionDemo begin");final long tick = System.currentTimeMillis();final List<CompletableFuture<?>> sequentCfs = new ArrayList<>();CompletableFuture<Integer> incomplete = new CompletableFuture<>();CompletableFuture<?> cf = incomplete.orTimeout(100, TimeUnit.MILLISECONDS).handle((v, ex) -> {logWithTimeAndThread("[1] timout");sleep(1000);return null;});sequentCfs.add(cf);cf = incomplete.orTimeout(100, TimeUnit.MILLISECONDS).handle((v, ex) -> {logWithTimeAndThread("[2] timout");sleep(1000);return null;});sequentCfs.add(cf);cf = incomplete.orTimeout(100, TimeUnit.MILLISECONDS).handle((v, ex) -> {logWithTimeAndThread("[3] timout");sleep(1000);return null;});sequentCfs.add(cf);CompletableFuture.allOf(sequentCfs.toArray(CompletableFuture[]::new)).join();logWithTimeAndThread("dysfunctionDemo end in " + (System.currentTimeMillis() - tick) + "ms");}private static void cffuOrTimeoutFixDysfunctionDemo() {logWithTimeAndThread("cffuOrTimeoutFixDysfunctionDemo begin");final long tick = System.currentTimeMillis();final List<CompletableFuture<?>> sequentCfs = new ArrayList<>();CompletableFuture<Integer> incomplete = new CompletableFuture<>();CompletableFuture<?> cf = CompletableFutureUtils.cffuOrTimeout(incomplete, 100, TimeUnit.MILLISECONDS).handle((v, ex) -> {logWithTimeAndThread("[1] timout");sleep(1000);return null;});sequentCfs.add(cf);cf = CompletableFutureUtils.cffuOrTimeout(incomplete, 100, TimeUnit.MILLISECONDS).handle((v, ex) -> {logWithTimeAndThread("[2] timout");sleep(1000);return null;});sequentCfs.add(cf);cf = CompletableFutureUtils.cffuOrTimeout(incomplete, 100, TimeUnit.MILLISECONDS).handle((v, ex) -> {logWithTimeAndThread("[3] timout");sleep(1000);return null;});sequentCfs.add(cf);CompletableFuture.allOf(sequentCfs.toArray(CompletableFuture[]::new)).join();logWithTimeAndThread("cffuOrTimeoutFixDysfunctionDemo end in " + (System.currentTimeMillis() - tick) + "ms");}private static void logWithTimeAndThread(String msg) {System.out.printf("%tF %<tT.%<tL [%s] %s%n",System.currentTimeMillis(), Thread.currentThread().getName(), msg);}
}

执行结果如下：

代码思路是这样的：有3个运行1秒的任务，在超时之后运行，不切线程池（都在 Delayer 线程运行），运行了3秒，不能在设置100ms的超时后运行，因为单线程排队了。handle 方法传入的回调函数在 Delayer 线程中执行了。

示例代码中解决超时线程延迟执行的方法是使用CFFU提供的安全 timeout 方法，本文后面会分析相关源码。

2. 问题分析

为什么handle方法里的回调会在 CompletableFutureDelayScheduler 中执行？

// 这里的代码逐步深入到调用栈内部
public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {return uniHandleStage(null, fn);
}private <V> CompletableFuture<V> uniHandleStage(Executor e, BiFunction<? super T, Throwable, ? extends V> f) {if (f == null) throw new NullPointerException();CompletableFuture<V> d = newIncompleteFuture();Object r;if ((r = result) == null)// 加入回调栈中后续再执行unipush(new UniHandle<T,V>(e, d, this, f));else if (e == null)// 有结果，直接执行d.uniHandle(r, f, null);else {try {e.execute(new UniHandle<T,V>(null, d, this, f));} catch (Throwable ex) {d.result = encodeThrowable(ex);}}return d;
}final <S> boolean uniHandle(Object r,BiFunction<? super S, Throwable, ? extends T> f,UniHandle<S,T> c) {S s; Throwable x;if (result == null) {try {// 此次调用中 c 为空，无需关注UniHandle，甚至不需要知道UniHandle的具体职责if (c != null && !c.claim())return false;if (r instanceof AltResult) {x = ((AltResult)r).ex;s = null;} else {x = null;@SuppressWarnings("unchecked") S ss = (S) r;s = ss;}// 执行回调completeValue(f.apply(s, x));} catch (Throwable ex) {completeThrowable(ex);}}return true;}

我们把出现问题的原因简单总结一下：

CompletionStage 中不带 async 的方法可能会在不同的线程中执行。一般情况下，如果CF的结果已经计算出来，后续的回调在调用线程中执行，如果结果没有计算出来，后续的回调在上一步计算的线程中执行。

以下是一个简化的代码示例：

@Slf4j
public class TimeoutBugDemo {public static void main(String[] args) {new CompletableFuture<Integer>().orTimeout(1, TimeUnit.SECONDS).handle((v, ex) -> {log.info("v: {}", v, ex);return -1;}).join();}
}

handle 方法传入的回调方法会在delayer线程中执行，从执行日志看也确实如此：

Task :TimeoutBugDemo.main()
11:58:53.465 [CompletableFutureDelayScheduler] INFO com.example.demo.cftimeout.TimeoutBugDemo -- v: null
java.util.concurrent.TimeoutException: null
at java.base/java.util.concurrent.CompletableFuture$Timeout.run(CompletableFuture.java:2920)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)

3. CFFU 是如何解决线程传导的？

// CFFU 代码实现
public static <C extends CompletableFuture<?>> C cffuOrTimeout(C cfThis, Executor executorWhenTimeout, long timeout, TimeUnit unit) {requireNonNull(cfThis, "cfThis is null");requireNonNull(executorWhenTimeout, "executorWhenTimeout is null");requireNonNull(unit, "unit is null");return hopExecutorIfAtCfDelayerThread(orTimeout(cfThis, timeout, unit), executorWhenTimeout);
}// 核心实现代码
private static <C extends CompletableFuture<?>> C hopExecutorIfAtCfDelayerThread(C cf, Executor executor) {CompletableFuture<Object> ret = newIncompleteFuture(cf);// use `cf.handle` method(instead of `cf.whenComplete`) and return null in order to// prevent reporting the handled exception argument of this `action` at subsequent `exceptionally`cf.handle((v, ex) -> {if (!atCfDelayerThread()) completeCf(ret, v, ex);// 使用 executor 后，CF的后续回调操作就不会在Dalayer 线程中执行了else executor.execute(() -> completeCf(ret, v, ex));return null;}).exceptionally(ex -> reportUncaughtException("handle of executor hop", ex));return (C) ret;
}private static void completeCf(CompletableFuture<Object> cf, Object value, @Nullable Throwable ex) {try {// 写入到新CF中if (ex == null) cf.complete(value);else cf.completeExceptionally(ex);} catch (Throwable t) {if (ex != null) t.addSuppressed(ex);reportUncaughtException("completeCf", t);throw t; // rethrow exception, report to caller}
}

基本思路将结果写入到新的 CompletableFuture 中，为了避免后续回调使用 Delayer 线程，改用新增的线程，保证线程传导的安全性。

提示：有时我们需要关注链式调用返回的是新值还是原有对象，比如 CompletableFuture#orTimeout 返回的是当前对象this, CFFU中返回的是新的 CompletableFuture。

4. 最佳实践的启示

1. 使用优秀的 CompletableFuture 类库： CFFU，避免编程出错，减轻开发负担。
2. 可参考我在《深入理解 Future, CompletableFuture, ListenableFuture，回调机制》一文中所讲的，如果使用CompletableFuture，应该尽量显示使用async*方法，同时显式传入执行器executor参数。
3. 改为使用 Guava 中的 ListenableFuture。