CompletableFuture 超时了，后台任务为什么还在跑

假设一个导出接口最多等两秒，超过时间就告诉用户“稍后再试”。接口日志按时结束了，线程池却没有空下来，几十秒后甚至还能看到“导出完成”。

于是又在超时分支里补了一句 future.cancel(true)。这次 isCancelled() 也变成了 true，但后台工作照样继续。

问题出在这里：等待结果、标记结果和执行工作，是三个不同的过程。它们之间需要明确的连接，方法名里有一个 cancel，并不代表所有过程都会一起结束。

这篇只讨论这个边界。前面的导出是示例场景；文中的对照代码在本地 Java 8u504 上实际运行过，只修改内存标记，不访问网络、数据库或业务文件。

一、先把三个“结束”分开

一次异步调用里，至少有三件事需要分别回答：

get(timeout, unit) 约束的是这一次等待。等待超时会抛出 TimeoutException，并不因此自动调用取消方法。另一个线程仍有可能继续执行任务并完成结果。(Future 文档)

因此，下面这段代码只能说明调用方不再等了：

try {
    return result.get(2, TimeUnit.SECONDS);
} catch (TimeoutException ex) {
    return fallback;
}

它没有说明工作线程收到了什么指令，更没有说明下游请求已经停止。

另一个常见误读是 isDone()。取消也属于 Future 的完成状态；把它作为“工作线程已经退出”的证据，会把两个不同的生命周期混起来。

二、一个能直接运行的对照实验

实验分别通过两种入口提交同一段工作：

• CompletableFuture.runAsync(work, pool)；
• 普通单线程执行器的 pool.submit(work)。

为了避免用睡眠猜执行顺序，代码用三个 CountDownLatch 控制现场：started 确认工作已进入，release 阻止工作继续，exited 确认它已走到退出位置。await() 是可中断等待，收到中断会抛出 InterruptedException。(CountDownLatch 文档)

这点很重要：实验先确认任务已经开始，再取消。 它不拿“任务还没被调度”这种情况来替代运行中取消。

保存为 FutureCancellationLab.java：

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

public class FutureCancellationLab {
    private static void await(CountDownLatch latch) throws InterruptedException {
        if (!latch.await(5, TimeUnit.SECONDS)) {
            throw new AssertionError("Lab timed out");
        }
    }

    private static void runCase(boolean completable) throws Exception {
        String name = completable ? "CompletableFuture" : "Executor.submit";
        ExecutorService pool = Executors.newSingleThreadExecutor();
        CountDownLatch started = new CountDownLatch(1);
        CountDownLatch release = new CountDownLatch(1);
        CountDownLatch exited = new CountDownLatch(1);
        AtomicBoolean interrupted = new AtomicBoolean();
        AtomicBoolean effect = new AtomicBoolean();

        Runnable work = () -> {
            started.countDown();
            try {
                release.await();
                effect.set(true); // Only an in-memory flag, no external writes.
            } catch (InterruptedException ex) {
                interrupted.set(true);
                Thread.currentThread().interrupt();
            } finally {
                exited.countDown();
            }
        };

        try {
            Future<?> result = completable
                    ? CompletableFuture.runAsync(work, pool)
                    : pool.submit(work);
            await(started);

            try {
                result.get(1, TimeUnit.MILLISECONDS);
                throw new AssertionError("Expected a waiting timeout");
            } catch (TimeoutException expected) {
                System.out.println(name + " after get timeout: done=" + result.isDone());
            }

            if (!result.cancel(true)) {
                throw new AssertionError("Expected cancellation to succeed");
            }
            System.out.println(name + " after cancel: cancelled=" + result.isCancelled()
                    + ", done=" + result.isDone());

            release.countDown();
            await(exited);
            System.out.println(name + " worker: interrupted=" + interrupted.get()
                    + ", effect=" + effect.get());
            if (interrupted.get() == completable || effect.get() != completable) {
                throw new AssertionError("Unexpected worker outcome");
            }
        } finally {
            release.countDown();
            pool.shutdownNow();
            if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
                throw new AssertionError("Worker did not exit");
            }
        }
    }

    public static void main(String[] args) throws Exception {
        System.out.println("java.version=" + System.getProperty("java.version"));
        runCase(true);
        runCase(false);
    }
}

在已经配置好 JDK 的终端运行：

javac -encoding UTF-8 FutureCancellationLab.java
java FutureCancellationLab

本次实际输出：

java.version=1.8.0_504
CompletableFuture after get timeout: done=false
CompletableFuture after cancel: cancelled=true, done=true
CompletableFuture worker: interrupted=false, effect=true
Executor.submit after get timeout: done=false
Executor.submit after cancel: cancelled=true, done=true
Executor.submit worker: interrupted=true, effect=false

两种结果都显示 cancelled=true, done=true，实际工作却走了不同路径。

CompletableFuture 那组没有收到中断。主线程打开 release 之后，工作继续执行，把 effect 设成了 true。

普通执行器那组在等待处收到中断，进入异常分支，因此没有执行后面的标记操作。

这里的 effect 只是布尔变量，用来观察“取消以后，工作还能不能走到下一步”。实验没有真的导出文件，更没有发起任何外部操作。两组执行器也都在 finally 中关闭，并检查线程已经退出。

三、为什么同一个 cancel(true)，行为却不同

CompletableFuture 的 Java 8 文档明确说明：它不像 FutureTask 那样直接控制引起结果完成的计算；取消被处理成一种异常完成。mayInterruptIfRunning 参数在这个实现中不用于控制执行。(CompletableFuture 文档)

所以，对实验里已经开始执行的 work 而言，下面这句会改变结果对象的状态，却不会替它中断工作线程：

result.cancel(true);

普通执行器这一侧使用了另一条路径。AbstractExecutorService 的默认 submit 实现通过 newTaskFor 包装工作，默认包装类型是 FutureTask；子类可以覆盖这一行为。(AbstractExecutorService 文档)

FutureTask 对运行中任务执行 cancel(true) 时，会尝试通过中断执行线程来停止任务。注意这里仍然是“尝试”，不是把任意一段 Java 代码强行终止。(FutureTask 文档)

因此，看到 Future<?> 这个变量类型还不够，还应查清它由谁创建、取消控制的是哪一个对象、任务现在是否已经开始。也不能把普通线程池这次实验的行为，直接推广到所有第三方 Future 实现。

四、换成 orTimeout，也要继续追问工作由谁停止

Java 9 开始增加了 orTimeout 和 completeOnTimeout。例如，下面的代码需要 Java 9 或以上版本，不能放进前面的 Java 8 实验直接编译：

CompletableFuture<Result> result =
        CompletableFuture.supplyAsync(this::load, executor);

result.orTimeout(2, TimeUnit.SECONDS);

orTimeout 在期限到达时，尝试把当前这个 CompletableFuture 异常完成；completeOnTimeout 则尝试用指定值完成它。它们返回的也是当前对象，不是自动创建一个隔离的结果副本。(Java 17 API 文档)

这会改变依赖这个结果的后续处理，但不能据此认定 load() 已经退出。

如果同一个 Future 被多个调用方共享，还应确认给其中一次等待设置期限，是否允许改变大家共享的结果状态。只想限制某个调用方等多久，与要让整个共享计算进入超时状态，是两种需求。

这部分依据 API 文档解释，本文实测输出来自上一节的 Java 8 程序，二者的版本范围分别标明。

五、中断是一条需要接收方配合的退出路径

实验中，普通线程池的工作能够退出，是因为 release.await() 响应了中断，代码又在异常分支结束了执行。

假如工作吞掉异常后继续处理，取消请求并不能替它补上退出逻辑：

try {
    blockingStep();
} catch (InterruptedException ex) {
    // 只记一条日志，然后继续往下执行
}
writeResult();

常见的处理方式是向上抛出 InterruptedException，或者在不能继续抛出时恢复中断标记，并离开当前任务。恢复标记只是保留中断信息，不会自动替代码执行 return。

对不进行可中断等待的计算循环，可以在合适的边界检查 Thread.currentThread().isInterrupted()，再走资源清理和退出路径。Thread.interrupted() 会清除当前线程的中断标记，isInterrupted() 则不会；不要把两个检查混用。(Thread 文档)

阻塞 I/O、数据库驱动、HTTP 客户端各有自己的行为和取消方式。不能因为 CountDownLatch.await() 在实验里退出了，就认为任意网络读取也会按同样方式及时返回。

同样，shutdownNow() 也只是尽力停止运行中的任务，不保证忽略中断的工作能够结束。应用共享的执行器有自己的生命周期，不能在每个请求超时后把整个线程池关闭。(ExecutorService 文档)

六、需要取消底层任务，就保留底层任务的控制入口

如果业务使用普通线程池执行可协作取消的阻塞工作，一种做法是保留 submit() 返回的 Future，在等待超时或等待线程被中断时，把取消请求传给它。

下面的辅助类演示这层关系：等待预算从提交前开始计算，排队时间也算在里面；等待失败后请求取消，不把取消成功当成任务已经退出。

保存为 WithinBudget.java：

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public final class WithinBudget {
    public static <T> T call(ExecutorService executor, Callable<T> action,
                             long budget, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
        if (budget <= 0) {
            throw new IllegalArgumentException("budget must be positive");
        }
        long budgetNanos = unit.toNanos(budget);
        long started = System.nanoTime();
        Future<T> task = executor.submit(action);
        try {
            long remaining = budgetNanos - (System.nanoTime() - started);
            if (remaining <= 0) {
                throw new TimeoutException("No waiting budget remains");
            }
            return task.get(remaining, TimeUnit.NANOSECONDS);
        } catch (TimeoutException ex) {
            task.cancel(true);
            throw ex;
        } catch (InterruptedException ex) {
            task.cancel(true);
            Thread.currentThread().interrupt();
            throw ex;
        }
    }
}

例如，已有一个由应用统一管理的执行器时：

Result result = WithinBudget.call(
        executor, this::load, 2, TimeUnit.SECONDS);

这个辅助类已在本地 Java 8 上验证：正常返回结果、等待超时，以及可中断任务收到取消后退出。它没有覆盖真实 HTTP 客户端或数据库驱动的取消行为。

使用前仍有几个前提：

1. 提交策略不会把耗时工作转交给当前请求线程运行。 例如，线程池饱和后使用 CallerRunsPolicy，submit() 本身就可能同步执行工作，直到它结束才返回；等拿到 Future 再检查预算，已经无法限制这段耗时。这里应使用符合业务要求的有界队列和明确拒绝策略。(ThreadPoolExecutor 文档)
2. 底层调用也有自己的时限和释放方式。 连接、读取、总调用时限分别约束什么，要按实际客户端确认；只在调用外面包一层 Future，不能替代这些配置。
3. 各阶段消耗同一份剩余预算。 排队已经花掉的时间不能在调用下游时重新发一份；重试前也要重新计算剩余量。

这里用 System.nanoTime() 的差值测量同一 JVM 内的经过时间。它不表示日期时间，也不能把这个值直接传给另一台机器当作通用时间戳。(System 文档)

如果对外还需要 CompletableFuture 来组合后续处理，可以另外维护结果对象；但任务句柄、结果对象与底层 I/O 的取消关系，仍需要明确设计和验证，不能只返回一个新 CompletableFuture 就认为传播已经建立。

七、Future 完成以后，不要过早宣布资源已经归还

这个区别还会影响并发控制。

假设一个系统规定“同一时刻最多执行十个导出”，在提交时扣减名额，又在 whenComplete 里归还名额。如果结果对象先超时完成，实际导出还没退出，名额就可能提前回到池子里。新任务再进入，实际运行量便超过了原来的十个。

这不是 CompletableFuture 必然造成的问题，而是把“结果已经结束”当成“执行已经结束”的记账错误。前面的实验已经展示了两件事可以分开发生。

应先明确名额约束的是等待请求数、已提交任务数，还是实际持有资源的工作数。不同名额的归还时机不一样，实际资源的释放应跟随真正持有它的执行路径。

设计时还要覆盖任务被拒绝、尚未执行就取消、执行中退出等分支。不能把所有归还都塞进工作体的 finally，却忘了某些工作根本不会开始；也不能在结果回调和工作退出时各归还一次。

此外，取消不会倒放已经发生的操作。如果导出文件已经生成，或请求已经被下游接收，本地 Future 进入取消状态，并不会自动删除文件或撤销远端处理。是否继续保存结果、按任务标识查询，或者执行清理，应由具体业务决定。

八、回归时，同时验证结果和工作现场

只断言接口在两秒内返回 TIMEOUT，测试还没有覆盖后台任务的生命周期。可以把回归拆成这些观察点：

这些是回归设计；本文实际运行验证的是完整对照程序和等待预算辅助类，不把表中所有场景都标成已实测。

排查线上问题时，建议给一次工作分配可关联的任务标识，分别记录提交、开始、调用方停止等待、发出取消、实际退出这几个时刻。超时日志与退出日志之间隔了多久，往往比一句 cancelled=true 更能说明问题。

以前的 CompletableFuture 设计与实践 讨论了结果组合。这次对照实验补上另一端：结果对象已经结束，并不意味着产生结果的工作已经结束。

把等待期限、取消传播、资源回收分别做清楚，超时才不只是界面上的两个字。