作者：京东科技 张天赐
前言
JDK 8 是一次重大的版本升级，新增了非常多的特性，其中之一便是 CompletableFuture。自此从 JDK 层面真正意义上的支持了基于事件的异步编程范式，弥补了 Future 的缺陷。
在我们的日常优化中，最常用手段便是多线程并行执行。这时候就会涉及到 CompletableFuture 的使用。
常见使用方式
下面举例一个常见场景。
假如我们有两个 RPC 远程调用服务，我们需要获取两个 RPC 的结果后，再进行后续逻辑处理。
1public static void main(String[] args) { 2 // 任务 A，耗时 2 秒 3 int resultA = compute(1); 4 // 任务 B，耗时 2 秒 5 int resultB = compute(2); 6 7 // 后续业务逻辑处理 8 System.out.println(resultA + resultB); 9}
可以预估到，串行执行最少耗时 4 秒，并且 B 任务并不依赖 A 任务结果。
对于这种场景，我们通常会选择并行的方式优化，Demo 代码如下：
1public static void main(String[] args) { 2 // 仅简单举例，在生产代码中可别这么写！ 3 4 // 统计耗时的函数 5 time(() -> { 6 CompletableFuture<Integer> result = Stream.of(1, 2) 7 // 创建异步任务 8 .map(x -> CompletableFuture.supplyAsync(() -> compute(x), executor)) 9 // 聚合 10 .reduce(CompletableFuture.completedFuture(0), (x, y) -> x.thenCombineAsync(y, Integer::sum, executor)); 11 12 // 等待结果 13 try { 14 System.out.println("结果：" + result.get()); 15 } catch (ExecutionException | InterruptedException e) { 16 System.err.println("任务执行异常"); 17 } 18 }); 19} 20 21输出： 22[async-1]: 任务执行开始：1 23[async-2]: 任务执行开始：2 24[async-1]: 任务执行完成：1 25[async-2]: 任务执行完成：2 26结果：3 27耗时：2 秒
可以看到耗时变成了 2 秒。
存在的问题
分析
看上去 CompletableFuture 现有功能可以满足我们诉求。但当我们引入一些现实常见情况时，一些潜在的不足便暴露出来了。
compute(x) 如果是一个根据入参查询用户某类型优惠券列表的任务，我们需要查询两种优惠券并组合在一起返回给上游。假如上游要求我们 2 秒内处理完毕并返回结果，但 compute(x) 耗时却在 0.5 秒 ~ 无穷大波动。这时候我们就需要把耗时过长的 compute(x) 任务结果放弃，仅处理在指定时间内完成的任务，尽可能保证服务可用。
那么以上代码的耗时由耗时最长的服务决定，无法满足现有诉求。通常我们会使用 get(long timeout, TimeUnit unit) 来指定获取结果的超时时间，并且我们会给 compute(x) 设置一个超时时间，达到后自动抛异常来中断任务。
1public static void main(String[] args) { 2 // 仅简单举例，在生产代码中可别这么写！ 3 4 // 统计耗时的函数 5 time(() -> { 6 List<CompletableFuture<Integer>> result = Stream.of(1, 2) 7 // 创建异步任务，compute(x) 超时抛出异常 8 .map(x -> CompletableFuture.supplyAsync(() -> compute(x), executor)) 9 .toList(); 10 11 // 等待结果 12 int res = 0; 13 for (CompletableFuture<Integer> future : result) { 14 try { 15 res += future.get(2, SECONDS); 16 } catch (ExecutionException | InterruptedException | TimeoutException e) { 17 System.err.println("任务执行异常或超时"); 18 } 19 } 20 21 System.out.println("结果：" + res); 22 }); 23} 24 25输出： 26[async-2]: 任务执行开始：2 27[async-1]: 任务执行开始：1 28[async-1]: 任务执行完成：1 29任务执行异常或超时 30结果：1 31耗时：2 秒
可以看到，只要我们能够给 compute(x) 设置一个超时时间将任务中断，结合 get、getNow 等获取结果的方式，就可以很好地管理整体耗时。
那么问题也就转变成了，如何给任务设置异步超时时间呢？
现有做法
当异步任务是一个 RPC 请求时，我们可以设置一个 JSF 超时，以达到异步超时效果。
当请求是一个 R2M 请求时，我们也可以控制 R2M 连接的最大超时时间来达到效果。
这么看好像我们都是在依赖三方中间件的能力来管理任务超时时间？那么就存在一个问题，中间件超时控制能力有限，如果异步任务是中间件 IO 操作 + 本地计算操作怎么办？
用 JSF 超时举一个具体的例子，反编译 JSF 的获取结果代码如下：
1public V get(long timeout, TimeUnit unit) throws InterruptedException { 2 // 配置的超时时间 3 timeout = unit.toMillis(timeout); 4 // 剩余等待时间 5 long remaintime = timeout - (this.sentTime - this.genTime); 6 if (remaintime <= 0L) { 7 if (this.isDone()) { 8 // 反序列化获取结果 9 return this.getNow(); 10 } 11 } else if (this.await(remaintime, TimeUnit.MILLISECONDS)) { 12 // 等待时间内任务完成，反序列化获取结果 13 return this.getNow(); 14 } 15 16 this.setDoneTime(); 17 // 超时抛出异常 18 throw this.clientTimeoutException(false); 19}
当这个任务刚好卡在超时边缘完成时，这个任务的耗时时间就变成了超时时间 + 获取结果时间。而获取结果（反序列化）作为纯本地计算操作，耗时长短受 CPU 影响较大。
某些 CPU 使用率高的情况下，就会出现异步任务没能触发抛出异常中断，导致我们无法准确控制超时时间。对上游来说，本次请求全部失败。
解决方式
JDK 9
这类问题非常常见，如大促场景，服务器 CPU 瞬间升高就会出现以上问题。
那么如何解决呢？其实 JDK 的开发大佬们早有研究。在 JDK 9，CompletableFuture 正式提供了 orTimeout、completeTimeout 方法，来准确实现异步超时控制。
1public CompletableFuture<T> orTimeout(long timeout, TimeUnit unit) { 2 if (unit == null) 3 throw new NullPointerException(); 4 if (result == null) 5 whenComplete(new Canceller(Delayer.delay(new Timeout(this), timeout, unit))); 6 return this; 7}
JDK 9 orTimeout 其实现原理是通过一个定时任务，在给定时间之后抛出异常。如果任务在指定时间内完成，则取消抛异常的操作。
以上代码我们按执行顺序来看下：
首先执行 new Timeout(this)。
1static final class Timeout implements Runnable { 2 final CompletableFuture<?> f; 3 Timeout(CompletableFuture<?> f) { this.f = f; } 4 public void run() { 5 if (f != null && !f.isDone()) 6 // 抛出超时异常 7 f.completeExceptionally(new TimeoutException()); 8 } 9}
通过源码可以看到，Timeout 是一个实现 Runnable 的类，run() 方法负责给传入的异步任务通过 completeExceptionally CAS 赋值异常，将任务标记为异常完成。
那么谁来触发这个 run() 方法呢？我们看下 Delayer 的实现。
1static final class Delayer { 2 static ScheduledFuture<?> delay(Runnable command, long delay, 3 TimeUnit unit) { 4 // 到时间触发 command 任务 5 return delayer.schedule(command, delay, unit); 6 } 7 8 static final class DaemonThreadFactory implements ThreadFactory { 9 public Thread newThread(Runnable r) { 10 Thread t = new Thread(r); 11 t.setDaemon(true); 12 t.setName("CompletableFutureDelayScheduler"); 13 return t; 14 } 15 } 16 17 static final ScheduledThreadPoolExecutor delayer; 18 static { 19 (delayer = new ScheduledThreadPoolExecutor( 20 1, new DaemonThreadFactory())). 21 setRemoveOnCancelPolicy(true); 22 } 23}
Delayer 其实就是一个单例定时调度器，Delayer.delay(new Timeout(this), timeout, unit) 通过 ScheduledThreadPoolExecutor 实现指定时间后触发 Timeout 的 run() 方法。
到这里就已经实现了超时抛出异常的操作。但当任务完成时，就没必要触发 Timeout 了。因此我们还需要实现一个取消逻辑。
1static final class Canceller implements BiConsumer<Object, Throwable> { 2 final Future<?> f; 3 Canceller(Future<?> f) { this.f = f; } 4 public void accept(Object ignore, Throwable ex) { 5 if (ex == null && f != null && !f.isDone()) 6 // 3 未触发抛异常任务则取消 7 f.cancel(false); 8 } 9}
当任务执行完成，或者任务执行异常时，我们也就没必要抛出超时异常了。因此我们可以把 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(异步任务, 超时时间, 时间单位);
工具类源码：
1package com.jd.jr.market.reduction.util; 2 3import com.jdpay.market.common.exception.UncheckedException; 4 5import java.util.concurrent.*; 6import java.util.function.BiConsumer; 7 8/** 9 * CompletableFuture 扩展工具 10 * 11 * @author zhangtianci7 12 */ 13public class CompletableFutureExpandUtils { 14 15 /** 16 * 如果在给定超时之前未完成，则异常完成此 CompletableFuture 并抛出 {@link TimeoutException} 。 17 * 18 * @param timeout 在出现 TimeoutException 异常完成之前等待多长时间，以 {@code unit} 为单位 19 * @param unit 一个 {@link TimeUnit}，结合 {@code timeout} 参数，表示给定粒度单位的持续时间 20 * @return 入参的 CompletableFuture 21 */ 22 public static <T> CompletableFuture<T> orTimeout(CompletableFuture<T> future, long timeout, TimeUnit unit) { 23 if (null == unit) { 24 throw new UncheckedException("时间的给定粒度不能为空"); 25 } 26 if (null == future) { 27 throw new UncheckedException("异步任务不能为空"); 28 } 29 if (future.isDone()) { 30 return future; 31 } 32 33 return future.whenComplete(new Canceller(Delayer.delay(new Timeout(future), timeout, unit))); 34 } 35 36 /** 37 * 超时时异常完成的操作 38 */ 39 static final class Timeout implements Runnable { 40 final CompletableFuture<?> future; 41 42 Timeout(CompletableFuture<?> future) { 43 this.future = future; 44 } 45 46 public void run() { 47 if (null != future && !future.isDone()) { 48 future.completeExceptionally(new TimeoutException()); 49 } 50 } 51 } 52 53 /** 54 * 取消不需要的超时的操作 55 */ 56 static final class Canceller implements BiConsumer<Object, Throwable> { 57 final Future<?> future; 58 59 Canceller(Future<?> future) { 60 this.future = future; 61 } 62 63 public void accept(Object ignore, Throwable ex) { 64 if (null == ex && null != future && !future.isDone()) { 65 future.cancel(false); 66 } 67 } 68 } 69 70 /** 71 * 单例延迟调度器，仅用于启动和取消任务，一个线程就足够 72 */ 73 static final class Delayer { 74 static ScheduledFuture<?> delay(Runnable command, long delay, TimeUnit unit) { 75 return delayer.schedule(command, delay, unit); 76 } 77 78 static final class DaemonThreadFactory implements ThreadFactory { 79 public Thread newThread(Runnable r) { 80 Thread t = new Thread(r); 81 t.setDaemon(true); 82 t.setName("CompletableFutureExpandUtilsDelayScheduler"); 83 return t; 84 } 85 } 86 87 static final ScheduledThreadPoolExecutor delayer; 88 89 static { 90 delayer = new ScheduledThreadPoolExecutor(1, new DaemonThreadFactory()); 91 delayer.setRemoveOnCancelPolicy(true); 92 } 93 } 94}