Java Completable Future异步超时实践探索

Tech

导读

JDK 8 是一次重大的版本升级，新增了非常多的特性，其中之一便是 CompletableFuture。自此从 JDK 层面真正意义上的支持了基于事件的异步编程范式，弥补了 Future 的缺陷。在日常优化中，最常用手段便是多线程并行执行。这时候就会涉及到 CompletableFuture 的使用。

public static void main(String[] args) {    // 任务 A，耗时 2 秒    int resultA = compute(1);    // 任务 B，耗时 2 秒    int resultB = compute(2);
    // 后续业务逻辑处理    System.out.println(resultA + resultB);}

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 秒。

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 秒

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

03

解决方式

3.1 JDK 9

这类问题非常常见，如电商大促场景，服务器 CPU 瞬间升高就会出现以上问题。

那么如何解决呢？其实 JDK 的开发大佬们早有研究。在 JDK 9，CompletableFuture 正式提供了 or Timeout、complete Time out 方法，来准确实现异步超时控制。

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 or Timeout 其实现原理是通过一个定时任务，在给定时间之后抛出异常。如果任务在指定时间内完成，则取消抛异常的操作。

以上代码按执行顺序来看下：首先执行 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 () 方法。

到这里就已经实现了超时抛出异常的操作。但当任务完成时，就没必要触发 Time out 了。因此还需要实现一个取消逻辑。

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。当的异步任务完成，并且定时超时任务未完成的时候，就是取消的时机。因此可以通过 when Complete (Bi Consumer<? super T, ? super Throwable> action) 来完成。

Canceller 就是一个 BiConsumer 的实现。其持有了 delayer. schedule  (command, delay, unit) 返回的定时超时任务，accept(Object ignore, Throwable ex) 实现了定时超时任务未完成后，执行 cancel(boolean mayInterruptIfRunning) 取消任务的操作。

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

04

总结

在 JDK 8 场景下，现有超时中断的做法依赖于任务本身的超时实现，当任务本身的超时失效，或者不够精确时，并没有很好的手段来中断任务。因此本文给出一种让 CompletableFuture 支持异步超时的实现方案实现思路，仅供读者参考。