39. orTimeout 和 completeOnTimeout：异步任务超时怎么处理
发布于 • 阅读量 0
39. orTimeout 和 completeOnTimeout：异步任务超时怎么处理
前面学习 Future 时，用过带超时时间的 get()：
future.get(30, TimeUnit.SECONDS);
它解决的是：
当前线程最多等待多久。
到了 CompletableFuture 这里，Java 还提供了两个更适合异步流程的超时方法：
orTimeout(...)
completeOnTimeout(...)
它们都能给异步任务设置超时时间，但超时后的处理方式不同：
orTimeout：超时后，让任务以异常状态完成。
completeOnTimeout：超时后，返回一个默认结果。
我现在这样记：
超时算失败，用 orTimeout。
超时给默认值，用 completeOnTimeout。
先说明版本要求
orTimeout() 和 completeOnTimeout() 是 Java 9 加入的方法。
如果项目使用 Java 8，就不能直接调用这两个方法，需要通过其他方式实现超时控制。
这套练习如果使用 Java 17 或更高版本，可以直接使用。
orTimeout：超时后抛出异常
基本写法如下：
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> {
sleep(5000);
return "PDF 处理成功";
})
.orTimeout(3, TimeUnit.SECONDS);
这里异步任务需要 5 秒完成，但超时时间只设置了 3 秒。
因此，3 秒以后，这个 CompletableFuture 会以异常状态完成。
调用：
future.join();
会抛出 CompletionException，其内部原因通常是：
TimeoutException
orTimeout 完整示例
新建类：
com.succos.completablefuture.OrTimeoutDemo
代码如下：
package com.succos.completablefuture;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class OrTimeoutDemo {
public static void main(String[] args) {
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName()
+ " 开始处理 PDF");
sleep(5000);
System.out.println(Thread.currentThread().getName()
+ " PDF 处理完成");
return "output/test-watermark.pdf";
})
.orTimeout(3, TimeUnit.SECONDS);
try {
String result = future.join();
System.out.println("处理结果：" + result);
} catch (CompletionException e) {
Throwable cause = getRootCause(e);
if (cause instanceof TimeoutException) {
System.out.println("PDF 处理超时");
} else {
System.out.println("PDF 处理失败："
+ cause.getMessage());
}
}
}
private static Throwable getRootCause(Throwable throwable) {
Throwable cause = throwable;
while (cause.getCause() != null) {
cause = cause.getCause();
}
return cause;
}
private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("任务被中断", e);
}
}
}
因为任务耗时 5 秒，而超时时间是 3 秒，所以最终会进入异常处理。
可能输出：
ForkJoinPool.commonPool-worker-1 开始处理 PDF
PDF 处理超时
orTimeout 不会直接返回默认结果
调用：
.orTimeout(3, TimeUnit.SECONDS)
只是让异步任务在超过指定时间后异常完成。
它不会自动返回：
处理失败；
默认路径；
空字符串；
失败结果对象。
如果想把超时异常转换成业务结果，还需要配合：
exceptionally(...)
或者：
handle(...)
例如：
CompletableFuture<PdfTaskResult> future = CompletableFuture
.supplyAsync(() -> {
String targetPath = addWatermark(file);
return PdfTaskResult.success(
file.getName(),
targetPath
);
}, pdfExecutor)
.orTimeout(30, TimeUnit.SECONDS)
.exceptionally(ex -> {
Throwable cause = getRootCause(ex);
if (cause instanceof TimeoutException) {
return PdfTaskResult.fail(
file.getName(),
"PDF 处理超时"
);
}
return PdfTaskResult.fail(
file.getName(),
cause.getMessage()
);
});
这样无论是普通异常还是超时异常，最后都能返回一个 PdfTaskResult。
completeOnTimeout：超时后返回默认结果
completeOnTimeout() 的基本写法如下：
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> {
sleep(5000);
return "PDF 处理成功";
})
.completeOnTimeout(
"PDF 处理超时",
3,
TimeUnit.SECONDS
);
这里任务同样需要 5 秒。
但超过 3 秒以后，不会抛超时异常，而是直接使用默认结果：
PDF 处理超时
所以：
String result = future.join();
可以正常拿到字符串。
completeOnTimeout 完整示例
新建类：
com.succos.completablefuture.CompleteOnTimeoutDemo
代码如下：
package com.succos.completablefuture;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
public class CompleteOnTimeoutDemo {
public static void main(String[] args) {
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> {
System.out.println(Thread.currentThread().getName()
+ " 开始处理 PDF");
sleep(5000);
System.out.println(Thread.currentThread().getName()
+ " PDF 处理完成");
return "output/test-watermark.pdf";
})
.completeOnTimeout(
"output/default.pdf",
3,
TimeUnit.SECONDS
);
String result = future.join();
System.out.println("最终结果：" + result);
}
private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("任务被中断", e);
}
}
}
任务超过 3 秒后，join() 会拿到：
output/default.pdf
而不是抛出 TimeoutException。
两个方法放在一起对比
| 方法 | 超时后的状态 | join() 的表现 |
适合场景 |
|---|---|---|---|
orTimeout |
异常完成 | 抛出 CompletionException |
超时必须明确算失败 |
completeOnTimeout |
正常完成 | 返回默认结果 | 超时可以使用降级结果 |
可以直接这样记：
orTimeout：超过时间，报错。
completeOnTimeout：超过时间，给默认值。
PDF 处理更适合哪一个
对于真正的 PDF 水印处理，我更倾向于使用：
orTimeout(...)
原因是 PDF 水印属于明确的业务任务。
如果处理没有在规定时间内完成，通常应该记录为：
任务超时；
处理失败；
需要重试；
不能生成正式下载地址。
我不太会给它返回一个假的默认路径：
output/default.pdf
因为用户可能把这个默认文件误认为自己处理后的 PDF。
所以更合理的写法是：
.orTimeout(30, TimeUnit.SECONDS)
.exceptionally(ex -> {
Throwable cause = getRootCause(ex);
if (cause instanceof TimeoutException) {
return PdfTaskResult.fail(
file.getName(),
"PDF 处理超过 30 秒"
);
}
return PdfTaskResult.fail(
file.getName(),
cause.getMessage()
);
})
超时后明确返回失败结果。
completeOnTimeout 更适合降级场景
completeOnTimeout() 更适合有合理默认值的场景。
例如：
查询用户头像超时，返回默认头像；
查询推荐内容超时，返回空列表；
读取非核心配置超时，使用默认配置；
查询辅助说明超时，返回默认提示。
这些场景即使超时，也可以继续完成主要业务。
比如：
CompletableFuture<String> avatarFuture =
queryUserAvatarAsync(userId)
.completeOnTimeout(
"/images/default-avatar.png",
2,
TimeUnit.SECONDS
);
如果头像服务超时，就显示默认头像。
这很合理。
但 PDF 水印超时后直接返回一个默认 PDF，往往不合理。
所以选择哪个方法，关键不是 API 本身，而是业务是否允许降级。
超时以后，原任务会自动停止吗
这是这一节最需要注意的地方。
无论使用：
orTimeout(...)
还是：
completeOnTimeout(...)
它们主要改变的是 CompletableFuture 的完成状态。
它们不等于强制停止底层任务。
也就是说：
调用方 3 秒后已经拿到超时异常或默认结果；
后台原来的任务仍然可能继续执行。
例如：
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> {
System.out.println("任务开始");
sleep(5000);
System.out.println("任务真正执行结束");
return "处理成功";
})
.completeOnTimeout(
"默认结果",
2,
TimeUnit.SECONDS
);
System.out.println(future.join());
sleep(5000);
可能输出：
任务开始
默认结果
任务真正执行结束
这说明超时结果已经返回，但原任务并没有因此立即停止。
超时控制和任务取消是两回事
我现在会这样区分：
orTimeout 和 completeOnTimeout：
控制 CompletableFuture 多久以后结束等待。
cancel(true)：
尝试中断底层任务。
它们不是一回事。
即使超时后再调用：
future.cancel(true);
也不保证底层业务马上停止。
任务能不能停下来，还是取决于：
任务代码是否检查中断状态；
阻塞方法是否响应中断；
底层 PDF 库是否支持中断；
文件写入是否已经开始。
所以不能把 orTimeout() 理解成“执行超过 30 秒就杀掉线程”。
它没有这个能力。
为什么底层任务可能继续执行
CompletableFuture 管理的是任务完成状态。
而真正的任务可能已经在线程池线程里运行：
CompletableFuture.supplyAsync(() -> {
return addWatermark(file);
}, pdfExecutor);
超时发生时，CompletableFuture 可以被标记为异常完成。
但执行 addWatermark(file) 的线程，可能仍在 PDFBox 内部处理页面。
如果底层代码没有响应中断，就会继续运行到结束。
所以超时只是告诉业务调用方：
这个结果已经来不及了，我不再正常等待它。
不代表后台工作一定停止。
PDF 超时后要防止半成品文件
PDF 水印处理通常会写输出文件。
如果任务超时，但后台还在写文件，可能出现：
接口已经返回处理超时；
输出目录里仍在生成文件；
用户重新提交后，两个任务写同一个路径；
生成的文件不完整；
半成品被当成正式文件下载。
所以输出文件设计要谨慎。
我更倾向于先写临时文件：
output/test-watermark.tmp.pdf
处理完全成功后，再改名为：
output/test-watermark.pdf
流程可以设计成：
1. 生成唯一临时文件名；
2. 把水印结果写入临时文件；
3. PDF 完整保存成功；
4. 把临时文件原子移动到正式路径；
5. 任务失败或超时时，清理临时文件。
这样至少不会轻易把半成品暴露给用户。
给 PDF 任务增加超时结果
可以定义一个结果对象：
static class PdfTaskResult {
private boolean success;
private boolean timeout;
private String fileName;
private String targetPath;
private String message;
}
超时时返回：
PdfTaskResult.timeout(
file.getName(),
"PDF 处理超过 30 秒"
)
普通失败时返回：
PdfTaskResult.fail(
file.getName(),
"PDF 文件损坏"
)
这样超时和普通失败就能区分开。
完整的 PDF 超时示例
新建类：
com.succos.completablefuture.PdfTimeoutDemo
代码如下：
package com.succos.completablefuture;
import java.io.File;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
public class PdfTimeoutDemo {
public static void main(String[] args) {
ThreadPoolExecutor pdfExecutor = new ThreadPoolExecutor(
3,
3,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory("pdf-worker"),
new ThreadPoolExecutor.CallerRunsPolicy()
);
File file = new File("input/test.pdf");
CompletableFuture<PdfTaskResult> future =
CompletableFuture
.supplyAsync(() -> {
String targetPath =
processPdf(file);
return PdfTaskResult.success(
file.getName(),
targetPath
);
}, pdfExecutor)
.orTimeout(
3,
TimeUnit.SECONDS
)
.exceptionally(ex -> {
Throwable cause =
getRootCause(ex);
if (cause instanceof TimeoutException) {
return PdfTaskResult.timeout(
file.getName(),
"PDF 处理超过 3 秒"
);
}
return PdfTaskResult.fail(
file.getName(),
cause.getMessage()
);
});
PdfTaskResult result = future.join();
System.out.println("--------------------------------");
System.out.println("文件名：" + result.getFileName());
System.out.println("是否成功：" + result.isSuccess());
System.out.println("是否超时：" + result.isTimeout());
System.out.println("输出路径：" + result.getTargetPath());
System.out.println("处理信息：" + result.getMessage());
pdfExecutor.shutdown();
}
private static String processPdf(File file) {
System.out.println(Thread.currentThread().getName()
+ " 开始处理："
+ file.getName());
sleep(5000);
String targetPath = "output/"
+ file.getName()
.replace(".pdf", "-watermark.pdf");
System.out.println(Thread.currentThread().getName()
+ " 处理完成："
+ targetPath);
return targetPath;
}
private static Throwable getRootCause(
Throwable throwable
) {
Throwable cause = throwable;
while (cause.getCause() != null) {
cause = cause.getCause();
}
return cause;
}
private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("任务被中断", e);
}
}
static class PdfTaskResult {
private boolean success;
private boolean timeout;
private String fileName;
private String targetPath;
private String message;
public static PdfTaskResult success(
String fileName,
String targetPath
) {
PdfTaskResult result = new PdfTaskResult();
result.success = true;
result.timeout = false;
result.fileName = fileName;
result.targetPath = targetPath;
result.message = "处理成功";
return result;
}
public static PdfTaskResult fail(
String fileName,
String message
) {
PdfTaskResult result = new PdfTaskResult();
result.success = false;
result.timeout = false;
result.fileName = fileName;
result.message = message;
return result;
}
public static PdfTaskResult timeout(
String fileName,
String message
) {
PdfTaskResult result = new PdfTaskResult();
result.success = false;
result.timeout = true;
result.fileName = fileName;
result.message = message;
return result;
}
public boolean isSuccess() {
return success;
}
public boolean isTimeout() {
return timeout;
}
public String getFileName() {
return fileName;
}
public String getTargetPath() {
return targetPath;
}
public String getMessage() {
return message;
}
}
static class NamedThreadFactory
implements ThreadFactory {
private final String prefix;
private final AtomicInteger threadNumber =
new AtomicInteger(1);
public NamedThreadFactory(String prefix) {
this.prefix = prefix;
}
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName(
prefix
+ "-"
+ threadNumber.getAndIncrement()
);
return thread;
}
}
}
这个示例中，PDF 处理模拟耗时 5 秒，但超时时间只有 3 秒。
所以最终结果会被标记为超时。
不过后台 processPdf() 是否立即停止，不能只依赖 orTimeout()。
批量 PDF 怎么给每个任务设置超时
如果有一批 PDF，可以在创建每个 CompletableFuture 时分别设置超时：
List<CompletableFuture<PdfTaskResult>> futureList =
new ArrayList<>();
for (File file : files) {
CompletableFuture<PdfTaskResult> future =
CompletableFuture
.supplyAsync(() -> {
String targetPath =
addWatermark(file);
return PdfTaskResult.success(
file.getName(),
targetPath
);
}, pdfExecutor)
.orTimeout(
30,
TimeUnit.SECONDS
)
.exceptionally(ex -> {
Throwable cause =
getRootCause(ex);
if (cause instanceof TimeoutException) {
return PdfTaskResult.timeout(
file.getName(),
"PDF 处理超时"
);
}
return PdfTaskResult.fail(
file.getName(),
cause.getMessage()
);
});
futureList.add(future);
}
然后统一等待：
CompletableFuture.allOf(
futureList.toArray(new CompletableFuture[0])
).join();
再收集结果：
List<PdfTaskResult> resultList = futureList
.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList());
因为每个任务都通过 exceptionally() 转成了结果对象，所以某个 PDF 超时不会直接让整个 allOf().join() 抛异常。
每个任务超时和整批任务超时不一样
这里还要区分两种需求。
第一种：
单个 PDF 最多处理 30 秒。
这种情况给每个任务分别加：
.orTimeout(30, TimeUnit.SECONDS)
第二种：
整批 PDF 最多等待 10 分钟。
这种情况可以给 allOf() 的结果设置超时：
CompletableFuture<Void> allFuture =
CompletableFuture.allOf(
futureList.toArray(
new CompletableFuture[0]
)
);
allFuture.orTimeout(
10,
TimeUnit.MINUTES
).join();
两者不一样：
单任务超时：限制每个文件的处理时间。
整批超时：限制整个批次最多等待多久。
真实项目里可能两种限制都需要。
整批超时也不会自动停止所有任务
即使写了：
allFuture.orTimeout(
10,
TimeUnit.MINUTES
).join();
超过 10 分钟后，allFuture 会异常完成。
但各个 PDF 子任务可能仍然继续执行。
如果希望超时后尝试取消未完成任务，可以在捕获超时后遍历：
for (CompletableFuture<PdfTaskResult> future
: futureList) {
if (!future.isDone()) {
future.cancel(true);
}
}
但仍然只是尝试取消。
不能保证所有底层任务立即停止。
completeOnTimeout 不要掩盖关键故障
completeOnTimeout() 使用起来很方便：
.completeOnTimeout(
defaultResult,
3,
TimeUnit.SECONDS
)
但也有一个风险：它可能把真正的性能问题隐藏起来。
例如查询数据库本来应该 100 毫秒完成，现在经常超过 3 秒。
如果每次都返回默认值，用户表面上还能看到页面，但系统问题一直没有暴露。
所以使用默认结果时，仍然应该记录超时日志和监控指标。
例如：
CompletableFuture<String> future =
queryConfigAsync()
.completeOnTimeout(
"默认配置",
3,
TimeUnit.SECONDS
)
.whenComplete((result, ex) -> {
// 这里只看最终状态，未必能直接判断是否用了超时默认值
});
如果业务需要准确知道是否发生过超时，更适合返回一个带状态的结果对象，而不是只返回普通字符串。
用结果对象表示降级
例如：
static class QueryResult<T> {
private boolean success;
private boolean degraded;
private T data;
private String message;
}
超时时返回：
QueryResult.degraded(
defaultValue,
"查询超时，已使用默认值"
)
这样业务能区分：
正常查询结果；
超时后的降级结果。
不会把两者混在一起。
orTimeout 和 exceptionally 的顺序
通常写成：
CompletableFuture
.supplyAsync(...)
.orTimeout(3, TimeUnit.SECONDS)
.exceptionally(ex -> {
return fallback;
});
顺序表示：
先给前面的异步任务增加超时限制；
如果任务异常或超时，再统一兜底。
如果把 exceptionally() 放在 orTimeout() 前面：
CompletableFuture
.supplyAsync(...)
.exceptionally(ex -> fallback)
.orTimeout(3, TimeUnit.SECONDS);
前面的普通业务异常可能已经被恢复了，但整条恢复后的流程仍然可能触发后面的超时。
方法顺序不同，含义也会变化。
我更常用：
.orTimeout(...)
.exceptionally(...)
先规定时间边界，再统一处理异常。
超时时间怎么定
超时时间不能随便写一个数字。
应该结合：
PDF 平均页数；
文件大小；
历史处理耗时；
机器 CPU 和内存；
磁盘读写速度；
任务排队时间；
业务能接受的等待时间。
例如，绝大多数 PDF 在 5 秒内完成，少量大文件需要 15 秒。
那超时时间可以先设置为 30 秒，再通过日志观察。
如果直接设置成 3 秒，可能大量正常文件都被误判为超时。
如果设置成 30 分钟，超时控制又失去了意义。
比较实际的做法是：
先记录真实耗时分布；
再根据 P95、P99 耗时设置边界；
对特殊大文件单独限制大小和页数。
学习项目里可以写 3 秒方便观察。
真实项目里要根据数据来定。
这一节小结
这一节我主要记住几点：
1. orTimeout 超时后让 CompletableFuture 异常完成；
2. completeOnTimeout 超时后返回一个默认结果；
3. 超时算业务失败时，更适合使用 orTimeout；
4. 业务允许降级时，可以使用 completeOnTimeout；
5. 两个方法都不等于强制停止底层任务；
6. PDF 超时后要考虑临时文件、半成品和状态清理；
7. 批量任务可以给每个 PDF 单独设置超时；
8. 单任务超时和整批任务超时是两种不同的限制；
9. 超时时间应该根据真实耗时和业务要求确定。
用一句话总结：
超时必须失败，用 orTimeout；超时允许降级，用 completeOnTimeout。
下一节开始整理 CompletableFuture 批量处理 PDF 的最终版本。
前面学习的自定义线程池、allOf、结果对象、异常处理、超时控制和日志记录，会在下一节合到一套完整代码里。
