44. ThreadPoolTaskExecutor 和 @Async 的使用方法
发布于 • 阅读量 0
44. ThreadPoolTaskExecutor 和 @Async 的使用方法
上一节使用的是 Java 原生线程池：
ThreadPoolExecutor
并且通过：
@Bean
把它交给 Spring 管理。
这种方式没有问题。
不过在 Spring Boot 项目中，还经常使用 Spring 提供的：
ThreadPoolTaskExecutor
它对 Java 原生的 ThreadPoolExecutor 做了一层封装，提供了更符合 Spring 使用习惯的配置方式，并且能够直接配合：
@Async
实现异步方法调用。
Spring 官方文档把 ThreadPoolTaskExecutor 作为常用的任务执行器实现。它底层仍然使用 ThreadPoolExecutor，同时能够参与 Spring 的生命周期管理。
这一节主要完成以下内容：
使用 ThreadPoolTaskExecutor 创建 PDF 线程池；
通过 @EnableAsync 开启异步方法支持；
通过 @Async 指定线程池；
让 PDF 方法在线程池中执行；
理解 @Async 的代理机制；
解决同类内部调用导致 @Async 失效的问题；
处理异步方法中的异常。
ThreadPoolTaskExecutor 是什么
ThreadPoolTaskExecutor 是 Spring 提供的线程池执行器。
它的完整类名是：
org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
它和 Java 原生线程池的关系可以简单理解为：
ThreadPoolExecutor：
Java 提供的真正线程池实现。
ThreadPoolTaskExecutor：
Spring 对 ThreadPoolExecutor 的封装。
使用原生线程池时，需要这样创建：
ThreadPoolExecutor executor =
new ThreadPoolExecutor(
3,
6,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory("pdf-worker"),
new ThreadPoolExecutor.AbortPolicy()
);
使用 ThreadPoolTaskExecutor 时，可以改成属性配置：
ThreadPoolTaskExecutor executor =
new ThreadPoolTaskExecutor();
executor.setCorePoolSize(3);
executor.setMaxPoolSize(6);
executor.setQueueCapacity(100);
executor.setKeepAliveSeconds(60);
executor.setThreadNamePrefix("pdf-worker-");
executor.setRejectedExecutionHandler(
new ThreadPoolExecutor.AbortPolicy()
);
两种方式表达的核心配置基本一致。
为什么 Spring 项目常用 ThreadPoolTaskExecutor
使用 ThreadPoolTaskExecutor 的主要原因不是它比 ThreadPoolExecutor 更强，而是它和 Spring 结合得更自然。
它可以：
作为 Spring Bean 管理；
参与应用启动和关闭生命周期；
直接配合 @Async 使用；
通过 Bean 名称选择线程池；
配置优雅关闭；
读取当前线程池状态；
在 Spring 配置类中集中维护。
所以在普通 Java 程序中，可以直接使用：
ThreadPoolExecutor
在 Spring Boot 项目中，也可以优先考虑：
ThreadPoolTaskExecutor
但无论使用哪一个，底层线程池的核心概念都没有变化：
核心线程数；
最大线程数；
任务队列；
线程存活时间；
线程工厂；
拒绝策略。
第一步：开启 Spring 异步方法支持
要使用：
@Async
首先要开启 Spring 的异步方法支持。
可以在配置类上添加：
@EnableAsync
例如新建：
com.succos.config.AsyncThreadPoolConfig
代码如下：
package com.succos.config;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
@Configuration
@EnableAsync
public class AsyncThreadPoolConfig {
}
@EnableAsync 的作用是：
开启 Spring 对 @Async 注解的识别和处理。
如果没有添加：
@EnableAsync
即使业务方法写了：
@Async
方法也可能仍然在调用线程中同步执行。
Spring 官方示例同样通过 @EnableAsync 开启异步方法功能。
可以把 @EnableAsync 放在启动类上吗
可以。
例如：
package com.succos;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;
@EnableAsync
@SpringBootApplication
public class PdfApplication {
public static void main(String[] args) {
SpringApplication.run(
PdfApplication.class,
args
);
}
}
这样也能开启异步方法支持。
不过为了让职责更清楚，我更倾向于把它放在线程池配置类中：
@Configuration
@EnableAsync
public class AsyncThreadPoolConfig {
}
这样看到配置类时就知道：
这个类负责异步执行相关配置。
第二步：创建 ThreadPoolTaskExecutor Bean
完整配置如下：
package com.succos.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.ThreadPoolExecutor;
@Configuration
@EnableAsync
public class AsyncThreadPoolConfig {
@Bean(name = "pdfTaskExecutor")
public ThreadPoolTaskExecutor pdfTaskExecutor() {
ThreadPoolTaskExecutor executor =
new ThreadPoolTaskExecutor();
executor.setCorePoolSize(3);
executor.setMaxPoolSize(6);
executor.setQueueCapacity(100);
executor.setKeepAliveSeconds(60);
executor.setThreadNamePrefix(
"pdf-worker-"
);
executor.setRejectedExecutionHandler(
new ThreadPoolExecutor.AbortPolicy()
);
executor.setWaitForTasksToCompleteOnShutdown(
true
);
executor.setAwaitTerminationSeconds(30);
return executor;
}
}
这里创建了一个名称为：
pdfTaskExecutor
的线程池 Bean。
后面使用：
@Async("pdfTaskExecutor")
就可以指定 PDF 方法使用这个线程池。
为什么没有调用 initialize
有些示例会在最后写：
executor.initialize();
例如：
@Bean
public ThreadPoolTaskExecutor pdfTaskExecutor() {
ThreadPoolTaskExecutor executor =
new ThreadPoolTaskExecutor();
executor.setCorePoolSize(3);
executor.initialize();
return executor;
}
如果 ThreadPoolTaskExecutor 作为 Spring Bean 交给容器管理，Spring 会执行它的初始化生命周期，所以配置方法中通常不需要再手动调用：
initialize()
直接返回配置好的对象即可。
如果是在 Spring 容器之外自己创建和使用 ThreadPoolTaskExecutor，才需要自己关注初始化和关闭。
ThreadPoolTaskExecutor 的核心参数
corePoolSize
executor.setCorePoolSize(3);
表示核心线程数为 3。
可以理解为：
正常情况下，最多先使用 3 个核心线程处理任务。
例如同时提交三个 PDF：
a.pdf；
b.pdf；
c.pdf。
可能分别由：
pdf-worker-1；
pdf-worker-2；
pdf-worker-3。
同时处理。
maxPoolSize
executor.setMaxPoolSize(6);
表示线程池最多可以创建 6 个工作线程。
但不是一提交第 4 个任务，就马上创建第 4 个线程。
在线程池使用有界队列时，通常是：
先创建核心线程；
核心线程都忙以后，任务进入队列；
队列装满以后，才继续创建非核心线程；
线程数达到最大线程数以后，再提交任务会触发拒绝策略。
当前配置是：
核心线程数：3；
最大线程数：6；
队列容量：100。
因此大致执行顺序是：
前 3 个任务由核心线程执行；
接下来的任务进入长度为 100 的队列；
队列满了以后，才可能把线程数从 3 增加到 6；
线程达到 6，并且队列也满了以后，触发拒绝策略。
所以：
maxPoolSize
是否容易发挥作用，和：
queueCapacity
有很大关系。
queueCapacity
executor.setQueueCapacity(100);
表示任务队列最多保存 100 个等待执行的任务。
这个队列控制的是：
当前线程都在忙时，还允许多少个任务排队。
队列不能盲目设置得特别大。
例如：
executor.setQueueCapacity(100000);
虽然能减少任务被拒绝的概率，但可能导致：
任务大量积压；
内存占用增加；
用户等待时间越来越长；
系统已经过载，但外部看不出来；
应用关闭时还有大量任务没有处理。
队列容量应该代表系统愿意承受的待处理任务数量。
keepAliveSeconds
executor.setKeepAliveSeconds(60);
表示超过核心线程数的非核心线程，在空闲一定时间后可以被回收。
例如线程池因为高峰期从 3 个线程扩展到 6 个线程。
高峰过去以后，多出来的 3 个非核心线程空闲超过 60 秒，就可以被销毁。
核心线程默认通常不会因为空闲而被回收。
如果希望核心线程也可以超时回收，可以设置：
executor.setAllowCoreThreadTimeOut(true);
但对于长期运行、持续接收 PDF 任务的应用，一般不需要急着开启。
threadNamePrefix
executor.setThreadNamePrefix(
"pdf-worker-"
);
设置线程名称前缀。
最终日志中可能出现：
pdf-worker-1
pdf-worker-2
pdf-worker-3
线程名称是排查异步问题的重要信息。
如果不设置，日志中可能只看到不够直观的默认线程名。
设置业务名称以后，可以快速判断：
PDF 任务是否进入了正确线程池；
@Async 是否真的生效；
任务有没有误跑到公共线程池；
不同业务线程池之间是否相互混用。
rejectedExecutionHandler
executor.setRejectedExecutionHandler(
new ThreadPoolExecutor.AbortPolicy()
);
表示线程池无法继续接收任务时，抛出拒绝异常。
常见拒绝策略包括：
AbortPolicy：
直接抛出 RejectedExecutionException。
CallerRunsPolicy：
由提交任务的线程执行任务。
DiscardPolicy：
直接丢弃任务，不抛异常。
DiscardOldestPolicy：
丢弃队列中等待最久的任务，再尝试提交新任务。
PDF 处理属于明确的业务任务。
如果任务提交失败，我希望能够明确知道，而不是悄悄丢失任务。
所以更适合使用：
AbortPolicy
然后在接口层或任务提交层处理拒绝异常。
设置应用关闭时的处理方式
配置中写了：
executor.setWaitForTasksToCompleteOnShutdown(
true
);
以及：
executor.setAwaitTerminationSeconds(30);
表示应用关闭时，希望给正在执行或排队的任务一定时间完成。
setAwaitTerminationSeconds(30) 表示 Spring 容器关闭过程中，最多等待线程池终止 30 秒。Spring 官方文档说明，这个等待时间可以让剩余任务继续使用其他仍由容器管理的资源。
但要注意：
等待 30 秒不代表所有 PDF 一定能完成；
超过等待时间后，应用仍然需要继续关闭；
长任务仍然应该通过任务状态和恢复机制管理。
对于一个真正的 PDF 后台任务系统，不能只依赖应用关闭时等待几十秒。
还需要考虑：
任务状态保存；
应用重启恢复；
临时文件清理；
重复任务处理；
任务幂等。
第三步：编写带 @Async 的 PDF Service
新建：
com.succos.service.PdfAsyncService
代码如下：
package com.succos.service;
import com.succos.dto.PdfTaskResult;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.io.File;
import java.util.concurrent.CompletableFuture;
@Service
public class PdfAsyncService {
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
long start =
System.currentTimeMillis();
try {
String targetPath =
processPdf(file);
long cost =
System.currentTimeMillis()
- start;
PdfTaskResult result =
PdfTaskResult.success(
file.getName(),
targetPath,
cost
);
return CompletableFuture
.completedFuture(result);
} catch (Exception e) {
long cost =
System.currentTimeMillis()
- start;
PdfTaskResult result =
PdfTaskResult.fail(
file.getName(),
getErrorMessage(e),
cost
);
return CompletableFuture
.completedFuture(result);
}
}
private String processPdf(File file) {
validateFile(file);
System.out.println(
Thread.currentThread().getName()
+ " 开始处理："
+ file.getName()
);
sleep(3000);
String targetPath =
"output/"
+ file.getName()
.replace(
".pdf",
"-watermark.pdf"
);
System.out.println(
Thread.currentThread().getName()
+ " 处理完成："
+ targetPath
);
return targetPath;
}
private void validateFile(File file) {
if (file == null) {
throw new IllegalArgumentException(
"PDF 文件不能为空"
);
}
if (!file.exists()) {
throw new IllegalArgumentException(
"PDF 文件不存在"
);
}
if (!file.isFile()) {
throw new IllegalArgumentException(
"PDF 路径不是文件"
);
}
if (!file.canRead()) {
throw new IllegalArgumentException(
"PDF 文件不可读取"
);
}
}
private void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(
"PDF 处理任务被中断",
e
);
}
}
private String getErrorMessage(
Throwable throwable
) {
Throwable cause = throwable;
while (cause.getCause() != null) {
cause = cause.getCause();
}
String message =
cause.getMessage();
if (message == null
|| message.isBlank()) {
return cause
.getClass()
.getSimpleName();
}
return message;
}
}
这里最关键的是：
@Async("pdfTaskExecutor")
它表示：
这个方法通过 Spring 异步机制执行；
使用名称为 pdfTaskExecutor 的线程池。
@Async("pdfTaskExecutor") 中的名称是什么
这里写的：
@Async("pdfTaskExecutor")
对应配置类中的 Bean 名称：
@Bean(name = "pdfTaskExecutor")
两个名称必须一致。
如果写成：
@Async("pdfExecutor")
但容器中只有：
pdfTaskExecutor
Spring 就找不到指定执行器。
所以我会保持名称统一：
@Bean(name = "pdfTaskExecutor")
对应：
@Async("pdfTaskExecutor")
调用 @Async 方法时发生了什么
假设另一个 Service 调用：
CompletableFuture<PdfTaskResult> future =
pdfAsyncService.processPdfAsync(file);
大致流程是：
1. 调用方调用 PdfAsyncService；
2. 实际调用先经过 Spring 生成的代理对象；
3. 代理发现方法上有 @Async；
4. 代理把方法任务提交到 pdfTaskExecutor；
5. 调用方快速拿到 CompletableFuture；
6. PDF 方法在 pdf-worker 线程中执行；
7. 方法完成后，CompletableFuture 得到结果。
所以 @Async 并不是 Java 语言本身的功能。
它依赖 Spring 的代理机制。
为什么方法里返回 completedFuture
方法的返回类型是：
CompletableFuture<PdfTaskResult>
但方法体已经由：
@Async
安排到线程池中执行了。
因此，在方法内部完成业务处理后，可以直接返回：
CompletableFuture.completedFuture(result)
例如：
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
PdfTaskResult result =
processPdf(file);
return CompletableFuture
.completedFuture(result);
}
这里的 completedFuture() 表示：
当前异步方法的业务已经执行结束；
把结果包装成一个已经完成的 CompletableFuture。
不要在 @Async 方法中重复 supplyAsync
下面这种写法通常没有必要：
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
return CompletableFuture.supplyAsync(
() -> processPdf(file),
pdfTaskExecutor
);
}
因为：
@Async 已经把方法放进了线程池；
方法内部又通过 supplyAsync 提交了一次任务。
这相当于：
第一次异步：调用 processPdfAsync；
第二次异步：执行 processPdf。
如果没有特殊需求，这只是重复调度。
更直接的写法是：
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
PdfTaskResult result =
processPdf(file);
return CompletableFuture
.completedFuture(result);
}
或者不使用 @Async，改为手动编排：
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
return CompletableFuture.supplyAsync(
() -> processPdf(file),
pdfTaskExecutor
);
}
两种方案选择一种即可。
@Async 和 supplyAsync 怎么选
使用 @Async
适合：
希望通过注解声明异步方法；
方法本身就是一个完整异步业务边界；
不需要在方法内部手动选择复杂执行流程；
团队已经统一使用 Spring 异步机制。
代码形式：
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
PdfTaskResult result =
processPdf(file);
return CompletableFuture
.completedFuture(result);
}
使用 CompletableFuture.supplyAsync
适合：
需要在一个方法中编排多个异步步骤；
需要动态选择执行器；
需要 thenCompose、thenCombine、allOf；
希望异步行为在代码中更加显式；
不希望依赖 Spring 代理调用。
代码形式：
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
return CompletableFuture.supplyAsync(
() -> processPdf(file),
pdfTaskExecutor
);
}
两种方式不是必须一起使用
我现在会这样判断：
单个完整业务方法异步化：
可以使用 @Async。
复杂 CompletableFuture 流程编排：
使用 supplyAsync、thenCompose、thenCombine 等方法。
不要认为：
@Async + supplyAsync
叠加以后会更异步。
大多数情况下只会增加一次不必要的任务提交。
第四步：调用异步 Service
新建一个任务调度 Service：
com.succos.service.PdfTaskService
代码如下：
package com.succos.service;
import com.succos.dto.PdfTaskResult;
import org.springframework.stereotype.Service;
import java.io.File;
import java.util.concurrent.CompletableFuture;
@Service
public class PdfTaskService {
private final PdfAsyncService pdfAsyncService;
public PdfTaskService(
PdfAsyncService pdfAsyncService
) {
this.pdfAsyncService =
pdfAsyncService;
}
public CompletableFuture<PdfTaskResult>
submitPdf(String fileName) {
File file =
new File(
"input",
fileName
);
return pdfAsyncService
.processPdfAsync(file);
}
}
这里要注意：
PdfTaskService 调用 PdfAsyncService；
是两个不同的 Spring Bean。
这一点和 @Async 的代理机制有关。
第五步：Controller 返回 CompletableFuture
Controller 可以这样写：
package com.succos.controller;
import com.succos.dto.PdfTaskResult;
import com.succos.service.PdfTaskService;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.CompletableFuture;
@RestController
@RequestMapping("/pdf")
public class PdfTaskController {
private final PdfTaskService pdfTaskService;
public PdfTaskController(
PdfTaskService pdfTaskService
) {
this.pdfTaskService =
pdfTaskService;
}
@PostMapping("/watermark")
public CompletableFuture<PdfTaskResult>
addWatermark(
@RequestParam String fileName
) {
return pdfTaskService
.submitPdf(fileName);
}
}
请求：
POST /pdf/watermark?fileName=test.pdf
日志可能显示：
pdf-worker-1 开始处理：test.pdf
pdf-worker-1 处理完成：output/test-watermark.pdf
这说明：
Controller 请求线程没有直接执行 PDF 任务；
PDF 任务进入了 pdfTaskExecutor。
@Async 最大的坑：同类内部调用会失效
这是使用 @Async 时最重要的问题之一。
假设这样写：
@Service
public class PdfTaskService {
public void submitPdf(File file) {
processPdfAsync(file);
}
@Async("pdfTaskExecutor")
public void processPdfAsync(File file) {
System.out.println(
Thread.currentThread().getName()
);
processPdf(file);
}
private void processPdf(File file) {
// 处理 PDF
}
}
在同一个类中：
submitPdf()
直接调用：
processPdfAsync()
这时 @Async 很可能不会生效。
方法可能仍然在原来的调用线程中执行。
为什么同类内部调用会失效
Spring 默认通过代理对象处理 @Async。
外部调用时：
其他 Bean
↓
Spring 代理对象
↓
发现 @Async
↓
提交线程池
↓
真正的业务方法
但同类内部调用相当于：
this.processPdfAsync(file);
调用没有经过 Spring 代理对象。
流程变成：
当前对象内部直接调用方法
↓
没有经过代理
↓
@Async 拦截逻辑没有执行
↓
同步执行
Spring 官方文档明确说明，默认的 @Async 处理模式基于代理，只有通过代理的调用才能被拦截；同一个类内部的本地调用无法被这种方式拦截。
正确做法：把异步方法拆到另一个 Bean
推荐结构是：
PdfTaskService：
负责业务调度。
PdfAsyncService：
负责异步执行。
例如：
@Service
public class PdfTaskService {
private final PdfAsyncService pdfAsyncService;
public PdfTaskService(
PdfAsyncService pdfAsyncService
) {
this.pdfAsyncService =
pdfAsyncService;
}
public CompletableFuture<PdfTaskResult>
submitPdf(File file) {
return pdfAsyncService
.processPdfAsync(file);
}
}
异步类：
@Service
public class PdfAsyncService {
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
PdfTaskResult result =
processPdf(file);
return CompletableFuture
.completedFuture(result);
}
}
因为调用发生在两个 Spring Bean 之间，所以会经过代理。
这是最清楚、也比较容易维护的方式。
不建议通过自己注入自己解决
有时会看到这种写法：
@Service
public class PdfTaskService {
private final PdfTaskService self;
public PdfTaskService(
PdfTaskService self
) {
this.self = self;
}
public void submitPdf(File file) {
self.processPdfAsync(file);
}
@Async
public void processPdfAsync(File file) {
// 处理 PDF
}
}
这种做法可能引入：
循环依赖；
代码难以理解；
代理对象和真实对象混淆；
测试困难。
也有人通过：
AopContext.currentProxy()
拿当前代理对象。
这会让业务代码和 Spring AOP 机制高度耦合。
所以更推荐：
把异步方法拆到独立 Service。
private 方法上的 @Async 通常不会达到预期
例如：
@Async("pdfTaskExecutor")
private void processPdfAsync(File file) {
}
然后在类内部调用：
processPdfAsync(file);
这种调用不会经过外部代理，因此通常不能实现预期的异步效果。
更合理的写法是：
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
}
并且由其他 Spring Bean 调用。
我会记住两个条件：
异步方法定义在 Spring Bean 中；
异步方法由另一个 Bean 通过代理调用。
@Async 方法可以返回什么
常见返回类型包括：
void；
Future<T>；
CompletableFuture<T>。
对于 PDF 任务，我更推荐：
CompletableFuture<PdfTaskResult>
因为调用方可以：
获取处理结果；
继续 thenApply；
继续 thenCompose；
增加 exceptionally；
等待多个任务；
统一收集结果。
例如：
CompletableFuture<PdfTaskResult> future =
pdfAsyncService
.processPdfAsync(file);
future.thenAccept(result -> {
System.out.println(
result.getMessage()
);
});
尽量少使用 void 返回异步方法
下面这种写法可以执行：
@Async("pdfTaskExecutor")
public void processPdfAsync(File file) {
processPdf(file);
}
但是调用方拿不到：
任务结果；
任务异常；
任务完成状态；
任务返回值。
如果方法抛出异常，也不能通过返回的 Future 获取。
Spring 为 void 类型的异步方法提供了：
AsyncUncaughtExceptionHandler
用来处理无法通过 Future 返回给调用方的未捕获异常。
因此，重要业务任务更适合返回：
CompletableFuture<PdfTaskResult>
而不是 void。
CompletableFuture 返回值怎么处理异常
异步方法可以直接让异常抛出：
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
String targetPath =
processPdf(file);
PdfTaskResult result =
PdfTaskResult.success(
file.getName(),
targetPath,
0
);
return CompletableFuture
.completedFuture(result);
}
如果 processPdf(file) 抛出异常，Spring 返回的 CompletableFuture 会异常完成。
调用方可以处理：
return pdfAsyncService
.processPdfAsync(file)
.exceptionally(ex -> {
return PdfTaskResult.fail(
file.getName(),
getErrorMessage(ex),
0
);
});
这样异常处理可以放在调用链中。
异常处理放在异步方法内部还是外部
放在异步方法内部
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
try {
String path =
processPdf(file);
return CompletableFuture.completedFuture(
PdfTaskResult.success(
file.getName(),
path,
0
)
);
} catch (Exception e) {
return CompletableFuture.completedFuture(
PdfTaskResult.fail(
file.getName(),
e.getMessage(),
0
)
);
}
}
优点是：
方法最终总能返回 PdfTaskResult；
批量任务比较容易收集结果。
缺点是：
异常已经被转成普通结果；
调用方不能再通过异常链统一处理。
放在调用链外部
异步方法只负责执行：
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
String path =
processPdf(file);
return CompletableFuture.completedFuture(
PdfTaskResult.success(
file.getName(),
path,
0
)
);
}
调用方统一处理：
return pdfAsyncService
.processPdfAsync(file)
.exceptionally(ex -> {
return PdfTaskResult.fail(
file.getName(),
getErrorMessage(ex),
0
);
});
优点是：
异步方法职责简单；
异常处理策略可以由不同调用方决定。
对于批量 PDF 项目，我更倾向于最终把异常转换成统一的：
PdfTaskResult
具体放在异步 Service 还是调度 Service，要根据项目职责划分决定。
void 异步方法怎么处理异常
假设确实有一个不需要返回值的异步方法：
@Async("pdfTaskExecutor")
public void clearTemporaryFile(File file) {
throw new RuntimeException(
"临时文件删除失败"
);
}
由于返回类型是：
void
调用方拿不到 CompletableFuture。
这时可以实现：
AsyncConfigurer
配置全局异步异常处理器。
例如：
package com.succos.config;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import java.lang.reflect.Method;
import java.util.Arrays;
@Configuration
@EnableAsync
public class AsyncExceptionConfig
implements AsyncConfigurer {
@Override
public AsyncUncaughtExceptionHandler
getAsyncUncaughtExceptionHandler() {
return new AsyncUncaughtExceptionHandler() {
@Override
public void handleUncaughtException(
Throwable ex,
Method method,
Object... params
) {
System.err.println(
"异步方法执行异常"
);
System.err.println(
"方法名："
+ method.getName()
);
System.err.println(
"参数："
+ Arrays.toString(params)
);
ex.printStackTrace();
}
};
}
}
这样 void 异步方法中未捕获的异常，可以由这个处理器记录。
不过对于关键 PDF 任务，仍然推荐返回：
CompletableFuture<PdfTaskResult>
这样异常、结果和状态都更容易管理。
@Async 默认使用哪个线程池
如果只写：
@Async
public void processPdfAsync(File file) {
}
没有指定：
@Async("pdfTaskExecutor")
Spring 会尝试选择默认异步执行器。
但当项目中存在多个线程池时，依赖默认选择容易造成混乱。
例如系统里有：
pdfTaskExecutor；
uploadTaskExecutor；
notifyTaskExecutor。
为了让代码更明确，我会直接写：
@Async("pdfTaskExecutor")
这样只看方法注解，就能知道它使用哪个线程池。
多个业务线程池的配置
完整配置可以写成：
package com.succos.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.ThreadPoolExecutor;
@Configuration
@EnableAsync
public class AsyncThreadPoolConfig {
@Bean(name = "pdfTaskExecutor")
public ThreadPoolTaskExecutor
pdfTaskExecutor() {
ThreadPoolTaskExecutor executor =
new ThreadPoolTaskExecutor();
executor.setCorePoolSize(3);
executor.setMaxPoolSize(6);
executor.setQueueCapacity(100);
executor.setKeepAliveSeconds(60);
executor.setThreadNamePrefix(
"pdf-worker-"
);
executor.setRejectedExecutionHandler(
new ThreadPoolExecutor.AbortPolicy()
);
executor.setWaitForTasksToCompleteOnShutdown(
true
);
executor.setAwaitTerminationSeconds(30);
return executor;
}
@Bean(name = "uploadTaskExecutor")
public ThreadPoolTaskExecutor
uploadTaskExecutor() {
ThreadPoolTaskExecutor executor =
new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(10);
executor.setQueueCapacity(200);
executor.setKeepAliveSeconds(60);
executor.setThreadNamePrefix(
"upload-worker-"
);
executor.setRejectedExecutionHandler(
new ThreadPoolExecutor.AbortPolicy()
);
executor.setWaitForTasksToCompleteOnShutdown(
true
);
executor.setAwaitTerminationSeconds(30);
return executor;
}
@Bean(name = "notifyTaskExecutor")
public ThreadPoolTaskExecutor
notifyTaskExecutor() {
ThreadPoolTaskExecutor executor =
new ThreadPoolTaskExecutor();
executor.setCorePoolSize(2);
executor.setMaxPoolSize(4);
executor.setQueueCapacity(500);
executor.setKeepAliveSeconds(60);
executor.setThreadNamePrefix(
"notify-worker-"
);
executor.setRejectedExecutionHandler(
new ThreadPoolExecutor.AbortPolicy()
);
executor.setWaitForTasksToCompleteOnShutdown(
true
);
executor.setAwaitTerminationSeconds(30);
return executor;
}
}
对应的业务方法：
@Async("pdfTaskExecutor")
public CompletableFuture<String>
processPdfAsync(File file) {
}
@Async("uploadTaskExecutor")
public CompletableFuture<String>
uploadAsync(String filePath) {
}
@Async("notifyTaskExecutor")
public CompletableFuture<Void>
notifyAsync(String userId) {
}
这样可以实现业务隔离。
@Async 和事务的关系
假设调用方法上有事务：
@Transactional
public void createTask() {
saveTask();
pdfAsyncService.processPdfAsync(file);
}
异步方法会在线程池中的另一个线程执行。
线程切换以后，原调用线程中的事务上下文通常不会自动跟着传过去。
因此不能简单认为：
外层事务回滚；
异步方法中的数据库操作也一定跟着回滚。
异步方法如果需要事务，应该根据业务需要在异步方法内部建立自己的事务边界。
例如：
@Async("pdfTaskExecutor")
@Transactional
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
// 独立异步事务
}
但还要考虑一个现实问题：
外层事务可能还没有提交；
异步线程就已经开始查询数据。
异步线程可能查询不到外层刚保存但尚未提交的数据。
因此，事务内启动异步任务时，要谨慎设计调用时机。
比较稳妥的方式可能是：
事务提交后再发布事件；
监听事务提交后的事件；
再启动异步任务。
这部分属于异步任务和事务协作的内容，后面可以单独展开。
@Async 不会让业务自动变快
给方法加上：
@Async
只是改变方法在哪个线程执行。
它并不会自动减少：
PDF 实际处理时间；
文件读写时间；
数据库执行时间；
网络上传时间。
例如一个 PDF 仍然需要 10 秒处理。
使用 @Async 后，它可能仍然需要 10 秒。
改变的是：
调用线程不必亲自执行这 10 秒任务；
任务交给线程池执行。
如果同时提交多个互不依赖的 PDF，线程池可以让它们并发处理，从而缩短整批任务总时间。
但线程数过多也可能导致：
CPU 争用；
内存压力；
磁盘 IO 争用；
上下文切换增加。
所以异步不是线程越多越好。
ThreadPoolTaskExecutor 获取运行状态
如果需要监控线程池，可以注入：
ThreadPoolTaskExecutor
例如：
package com.succos.service;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
@Service
public class PdfExecutorMonitorService {
private final ThreadPoolTaskExecutor
pdfTaskExecutor;
public PdfExecutorMonitorService(
@Qualifier("pdfTaskExecutor")
ThreadPoolTaskExecutor pdfTaskExecutor
) {
this.pdfTaskExecutor =
pdfTaskExecutor;
}
public void printStatus() {
int poolSize =
pdfTaskExecutor.getPoolSize();
int activeCount =
pdfTaskExecutor
.getActiveCount();
int queueSize =
pdfTaskExecutor
.getThreadPoolExecutor()
.getQueue()
.size();
long completedTaskCount =
pdfTaskExecutor
.getThreadPoolExecutor()
.getCompletedTaskCount();
System.out.println(
"当前线程数："
+ poolSize
);
System.out.println(
"活跃线程数："
+ activeCount
);
System.out.println(
"队列任务数："
+ queueSize
);
System.out.println(
"已完成任务数："
+ completedTaskCount
);
}
}
可以重点监控：
当前线程数；
活跃线程数；
任务队列长度；
已完成任务数；
任务拒绝次数。
如果长期出现：
活跃线程数达到最大值；
队列接近满载；
任务持续被拒绝。
说明线程池已经接近或超过当前承载能力。
使用 @Async 处理批量 PDF
异步 Service：
@Service
public class PdfAsyncService {
@Async("pdfTaskExecutor")
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
long start =
System.currentTimeMillis();
try {
String path =
processPdf(file);
long cost =
System.currentTimeMillis()
- start;
return CompletableFuture
.completedFuture(
PdfTaskResult.success(
file.getName(),
path,
cost
)
);
} catch (Exception e) {
long cost =
System.currentTimeMillis()
- start;
return CompletableFuture
.completedFuture(
PdfTaskResult.fail(
file.getName(),
e.getMessage(),
cost
)
);
}
}
}
批量调度 Service：
@Service
public class PdfBatchTaskService {
private final PdfAsyncService
pdfAsyncService;
public PdfBatchTaskService(
PdfAsyncService pdfAsyncService
) {
this.pdfAsyncService =
pdfAsyncService;
}
public CompletableFuture<List<PdfTaskResult>>
processBatch(List<File> fileList) {
List<CompletableFuture<PdfTaskResult>>
futureList = fileList
.stream()
.map(
pdfAsyncService
::processPdfAsync
)
.toList();
CompletableFuture<Void> allFuture =
CompletableFuture.allOf(
futureList.toArray(
new CompletableFuture[0]
)
);
return allFuture.thenApply(
unused -> futureList
.stream()
.map(
CompletableFuture::join
)
.toList()
);
}
}
这里：
每个 PDF 调用一次 @Async 方法；
每个调用返回 CompletableFuture；
所有任务进入同一个 pdfTaskExecutor；
allOf 等待所有 PDF 完成；
最后收集全部结果。
并发数量仍然由：
pdfTaskExecutor
控制。
不是由：
allOf
控制。
@Async 任务被拒绝时怎么办
线程池使用：
new ThreadPoolExecutor.AbortPolicy()
当线程池和队列都满时，任务提交可能抛出：
TaskRejectedException
或者底层的：
RejectedExecutionException
这类异常发生在任务提交阶段。
因此，调用异步方法时也要考虑任务是否成功提交。
例如可以在业务入口统一捕获并返回：
当前任务过多，请稍后重试。
也可以增加全局异常处理。
但不要简单改成：
DiscardPolicy
否则任务可能被直接丢弃，而业务系统没有任何记录。
对于 PDF 这类用户明确提交的任务，我更倾向于：
拒绝也要记录；
拒绝也要返回明确结果；
不能静默丢弃。
ThreadPoolTaskExecutor 和 ThreadPoolExecutor 怎么选
两者都能完成 PDF 任务。
更适合 ThreadPoolExecutor 的情况
普通 Java 项目；
不依赖 Spring；
需要完全使用 JDK 原生接口；
希望直接控制底层 ThreadPoolExecutor。
更适合 ThreadPoolTaskExecutor 的情况
Spring Boot 项目；
需要配合 @Async；
需要 Spring 管理生命周期；
需要统一配置多个任务执行器；
希望方便读取线程池状态。
在当前 Spring Boot PDF 项目中，我更倾向于使用：
ThreadPoolTaskExecutor
@Async 和 CompletableFuture 手动编排怎么选
简单异步业务边界
例如：
提交一个 PDF；
后台处理；
返回一个 CompletableFuture。
可以使用：
@Async("pdfTaskExecutor")
代码简单直观。
复杂异步任务链
例如：
添加水印；
上传文件；
查询用户；
合并用户和文件结果；
保存数据库；
发送通知。
更适合显式使用：
CompletableFuture.supplyAsync();
thenCompose();
thenCombine();
thenApply();
exceptionally();
因为流程关系更清楚。
常见错误总结
错误一：忘记添加 @EnableAsync
@Async
public void processPdf() {
}
但项目中没有：
@EnableAsync
异步注解不会按预期工作。
错误二：同类内部调用
public void submit() {
processAsync();
}
@Async
public void processAsync() {
}
同类内部调用没有经过 Spring 代理，异步可能失效。
错误三：异步类不是 Spring Bean
PdfAsyncService service =
new PdfAsyncService();
service.processPdfAsync(file);
自己 new 出来的对象不受 Spring 代理管理。
@Async 不会生效。
应该注入：
private final PdfAsyncService
pdfAsyncService;
错误四：@Async 中又重复 supplyAsync
@Async
public CompletableFuture<String>
processAsync() {
return CompletableFuture.supplyAsync(
() -> process()
);
}
没有特殊目的时，这是重复异步调度。
错误五：不指定业务线程池
@Async
public void processPdf() {
}
当项目存在多个线程池时，不明确指定执行器，容易造成线程池使用混乱。
更清楚的是：
@Async("pdfTaskExecutor")
错误六：关键任务使用 void 返回
@Async
public void processPdf() {
}
调用方无法方便获取任务结果和异常。
关键任务更适合返回：
CompletableFuture<PdfTaskResult>
错误七：把 @Async 当成后台任务系统
@Async 只是把方法提交到线程池。
它不会自动提供：
任务 ID；
任务持久化；
任务进度；
失败重试；
应用重启恢复；
分布式任务调度。
如果 PDF 任务需要执行几分钟，并且要求可靠完成，就不能只靠：
@Async
还需要完整的任务状态管理。
最终代码结构
这一节完成后，项目结构可以是：
src/main/java/com/succos
├── PdfApplication.java
├── config
│ └── AsyncThreadPoolConfig.java
├── controller
│ └── PdfTaskController.java
├── dto
│ └── PdfTaskResult.java
└── service
├── PdfTaskService.java
├── PdfAsyncService.java
└── PdfBatchTaskService.java
各类职责如下。
AsyncThreadPoolConfig
开启 @Async；
创建 ThreadPoolTaskExecutor；
设置线程数；
设置队列；
设置拒绝策略；
设置关闭方式。
PdfAsyncService
定义真正的 @Async 方法；
执行单个 PDF 处理；
返回 CompletableFuture。
PdfTaskService
负责普通任务提交；
调用独立的异步 Service；
避免同类内部调用。
PdfBatchTaskService
提交多个 PDF；
保存 CompletableFuture 列表；
通过 allOf 等待任务；
收集批量结果。
PdfTaskController
接收 HTTP 请求；
调用任务 Service；
返回异步结果。
这一节小结
这一节我主要记住几点：
1. ThreadPoolTaskExecutor 是 Spring 对 ThreadPoolExecutor 的封装；
2. 使用 @Async 前需要通过 @EnableAsync 开启异步支持；
3. @Async("pdfTaskExecutor") 可以指定具体业务线程池；
4. ThreadPoolTaskExecutor 可以由 Spring 管理初始化和销毁；
5. @Async 已经完成线程切换，方法内部通常不需要再重复 supplyAsync；
6. @Async 默认依赖 Spring 代理，同类内部调用不会经过代理；
7. 异步方法最好拆到独立的 Spring Service 中；
8. 关键业务任务更适合返回 CompletableFuture，而不是 void；
9. void 异步方法的未捕获异常可以通过 AsyncUncaughtExceptionHandler 处理；
10. @Async 只是异步执行工具，不是完整的可靠任务系统。
用一句话总结：
ThreadPoolTaskExecutor 负责提供 Spring 线程池，@Async 负责把经过代理的方法提交到指定线程池中执行。
下一节继续学习：
为什么 @Async 同类调用会失效；
Spring 代理对象到底是什么；
为什么从 Controller 调用可以异步，从本类调用却不行。
这会把前面学习过的 Spring AOP、代理对象和 @Async 串起来。
