43. Spring Boot 中把线程池定义成 Bean
发布于 • 阅读量 2
43. Spring Boot 中把线程池定义成 Bean
上一节已经明确了：
线程池应该跟着应用一起创建和销毁；
不能在每次业务调用时重新创建。
在 Spring Boot 项目中，比较合适的做法是把线程池交给 Spring 容器管理。
整体结构是：
配置类创建线程池 Bean；
Spring 容器保存线程池对象；
Service 通过构造方法注入线程池；
业务方法只负责提交任务；
应用关闭时由 Spring 销毁线程池。
这一节会完整实现：
@Configuration 配置类；
@Bean 创建线程池；
@Qualifier 区分线程池；
Service 注入线程池；
CompletableFuture 使用线程池；
应用关闭时安全销毁。
为什么要把线程池定义成 Bean
如果在线程池配置类中定义：
@Bean
public ThreadPoolExecutor pdfExecutor() {
return new ThreadPoolExecutor(...);
}
Spring 启动时会调用这个方法，创建线程池对象。
之后其他类只要注入这个 Bean，就能使用同一个线程池。
也就是说：
线程池只创建一次；
所有 PDF 任务共用一个线程池；
线程数量和队列容量统一控制；
应用关闭时统一销毁。
这样线程池就真正成为了应用级资源。
第一步：创建线程工厂
先准备一个自定义线程工厂，用来设置线程名称。
新建：
com.succos.config.NamedThreadFactory
代码如下：
package com.succos.config;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
public class NamedThreadFactory implements ThreadFactory {
private final String threadNamePrefix;
private final AtomicInteger threadNumber =
new AtomicInteger(1);
public NamedThreadFactory(
String threadNamePrefix
) {
this.threadNamePrefix =
threadNamePrefix;
}
@Override
public Thread newThread(Runnable runnable) {
Thread thread =
new Thread(runnable);
thread.setName(
threadNamePrefix
+ "-"
+ threadNumber.getAndIncrement()
);
thread.setDaemon(false);
thread.setUncaughtExceptionHandler(
(currentThread, throwable) -> {
System.err.println(
"线程发生未捕获异常，thread="
+ currentThread.getName()
);
throwable.printStackTrace();
}
);
return thread;
}
}
最终线程名称类似：
pdf-worker-1
pdf-worker-2
pdf-worker-3
看到日志时，就知道任务来自 PDF 线程池。
第二步：创建线程池配置类
新建：
com.succos.config.ThreadPoolConfig
代码如下：
package com.succos.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@Configuration
public class ThreadPoolConfig {
@Bean(
name = "pdfExecutor",
destroyMethod = "shutdown"
)
public ThreadPoolExecutor pdfExecutor() {
return new ThreadPoolExecutor(
3,
3,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory(
"pdf-worker"
),
new ThreadPoolExecutor.AbortPolicy()
);
}
}
这里最重要的是：
@Configuration
和：
@Bean(name = "pdfExecutor")
@Configuration 做了什么
@Configuration 表示：
这是一个 Spring 配置类；
类中可以定义需要交给 Spring 管理的 Bean。
Spring 启动时会扫描这个类。
发现：
@Bean
public ThreadPoolExecutor pdfExecutor()
以后，会调用这个方法创建线程池，并把返回的对象保存到 Spring 容器中。
因此，其他 Spring Bean 可以注入这个线程池。
@Bean 做了什么
@Bean 表示：
这个方法的返回值交给 Spring 管理。
例如：
@Bean(name = "pdfExecutor")
public ThreadPoolExecutor pdfExecutor() {
return new ThreadPoolExecutor(...);
}
Spring 管理的是方法返回的：
ThreadPoolExecutor
而不是配置类方法本身。
Bean 的名称是：
pdfExecutor
默认情况下，如果不写：
name = "pdfExecutor"
Bean 名称也会使用方法名：
pdfExecutor
所以这两种写法效果相近：
@Bean
public ThreadPoolExecutor pdfExecutor() {
return new ThreadPoolExecutor(...);
}
以及：
@Bean(name = "pdfExecutor")
public ThreadPoolExecutor createPdfExecutor() {
return new ThreadPoolExecutor(...);
}
第二种通过 name 明确指定 Bean 名称。
我更倾向于让方法名和 Bean 名称保持一致：
@Bean(name = "pdfExecutor")
public ThreadPoolExecutor pdfExecutor()
看起来更直观。
destroyMethod = "shutdown" 做了什么
配置中写了：
@Bean(
name = "pdfExecutor",
destroyMethod = "shutdown"
)
表示 Spring 容器关闭时，会调用线程池的：
shutdown()
这样应用停止时，线程池会进入正常关闭流程：
不再接收新任务；
正在执行的任务继续执行；
队列中的任务继续执行；
所有任务完成后，线程池终止。
业务 Service 中不需要手动调用：
pdfExecutor.shutdown();
线程池的关闭由 Spring 统一负责。
为什么业务方法不能调用 shutdown
假设 Service 中写了：
public void submitPdf(File file) {
pdfExecutor.execute(() -> {
processPdf(file);
});
pdfExecutor.shutdown();
}
第一次请求提交完成后，线程池就开始关闭。
第二次请求再调用：
pdfExecutor.execute(...)
就可能抛出：
RejectedExecutionException
因为已经关闭的线程池不能再接收任务。
所以业务代码只能：
提交任务；
查询线程池状态；
不能随意关闭公共线程池。
第三步：在 Service 中注入线程池
新建：
com.succos.service.PdfTaskService
代码如下：
package com.succos.service;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import java.io.File;
import java.util.concurrent.ThreadPoolExecutor;
@Service
public class PdfTaskService {
private final ThreadPoolExecutor pdfExecutor;
public PdfTaskService(
@Qualifier("pdfExecutor")
ThreadPoolExecutor pdfExecutor
) {
this.pdfExecutor = pdfExecutor;
}
public void submitPdf(File file) {
pdfExecutor.execute(() -> {
processPdf(file);
});
}
private void processPdf(File file) {
System.out.println(
Thread.currentThread().getName()
+ " 开始处理："
+ file.getName()
);
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(
"PDF 任务被中断",
e
);
}
System.out.println(
Thread.currentThread().getName()
+ " 处理完成："
+ file.getName()
);
}
}
这里使用的是构造方法注入。
为什么推荐构造方法注入
构造方法注入写法是：
private final ThreadPoolExecutor pdfExecutor;
public PdfTaskService(
@Qualifier("pdfExecutor")
ThreadPoolExecutor pdfExecutor
) {
this.pdfExecutor = pdfExecutor;
}
它有几个好处：
依赖关系清楚；
字段可以声明为 final；
对象创建时就必须提供依赖；
更方便写单元测试；
不依赖字段反射注入。
我不太建议写成：
@Autowired
private ThreadPoolExecutor pdfExecutor;
这种字段注入虽然代码少，但依赖关系不够明显。
构造方法注入更适合长期维护。
@Qualifier 有什么作用
现在配置类里只有一个 ThreadPoolExecutor Bean：
@Bean(name = "pdfExecutor")
这种情况下，Spring 通常可以完成注入。
但以后很可能还会增加：
uploadExecutor；
notifyExecutor；
reportExecutor。
它们的类型都是：
ThreadPoolExecutor
这时仅根据类型注入就无法判断应该使用哪一个。
例如：
public PdfTaskService(
ThreadPoolExecutor executor
) {
}
Spring 会发现容器里有多个 ThreadPoolExecutor：
pdfExecutor；
uploadExecutor；
notifyExecutor。
于是可能报错：
NoUniqueBeanDefinitionException
所以应该使用：
@Qualifier("pdfExecutor")
明确告诉 Spring：
我要注入名称为 pdfExecutor 的线程池。
多个线程池的配置方式
假设系统中还需要文件上传线程池，可以继续在配置类中定义：
@Bean(
name = "uploadExecutor",
destroyMethod = "shutdown"
)
public ThreadPoolExecutor uploadExecutor() {
return new ThreadPoolExecutor(
5,
10,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(200),
new NamedThreadFactory(
"upload-worker"
),
new ThreadPoolExecutor.AbortPolicy()
);
}
完整配置类如下：
package com.succos.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@Configuration
public class ThreadPoolConfig {
@Bean(
name = "pdfExecutor",
destroyMethod = "shutdown"
)
public ThreadPoolExecutor pdfExecutor() {
return new ThreadPoolExecutor(
3,
3,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory(
"pdf-worker"
),
new ThreadPoolExecutor.AbortPolicy()
);
}
@Bean(
name = "uploadExecutor",
destroyMethod = "shutdown"
)
public ThreadPoolExecutor uploadExecutor() {
return new ThreadPoolExecutor(
5,
10,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(200),
new NamedThreadFactory(
"upload-worker"
),
new ThreadPoolExecutor.AbortPolicy()
);
}
}
此时业务类必须明确指定线程池。
PDF Service：
public PdfTaskService(
@Qualifier("pdfExecutor")
ThreadPoolExecutor pdfExecutor
) {
this.pdfExecutor = pdfExecutor;
}
上传 Service：
public UploadService(
@Qualifier("uploadExecutor")
ThreadPoolExecutor uploadExecutor
) {
this.uploadExecutor = uploadExecutor;
}
这样不同业务使用不同线程池。
第四步：使用 CompletableFuture 提交 PDF 任务
前面的 submitPdf() 使用了：
pdfExecutor.execute(...)
如果任务需要返回结果，可以改成：
CompletableFuture.supplyAsync(...)
例如：
package com.succos.service;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import java.io.File;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
@Service
public class PdfTaskService {
private final ThreadPoolExecutor pdfExecutor;
public PdfTaskService(
@Qualifier("pdfExecutor")
ThreadPoolExecutor pdfExecutor
) {
this.pdfExecutor = pdfExecutor;
}
public CompletableFuture<String> processPdfAsync(
File file
) {
return CompletableFuture.supplyAsync(
() -> processPdf(file),
pdfExecutor
);
}
private String processPdf(File 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 sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(
"PDF 任务被中断",
e
);
}
}
}
这里最重要的是：
CompletableFuture.supplyAsync(
() -> processPdf(file),
pdfExecutor
)
第二个参数明确传入了 Spring 管理的：
pdfExecutor
所以 PDF 任务不会跑到默认公共线程池。
为什么一定要传入 pdfExecutor
如果写成：
CompletableFuture.supplyAsync(() -> {
return processPdf(file);
});
没有传执行器，就会默认使用：
ForkJoinPool.commonPool
线程名称可能是：
ForkJoinPool.commonPool-worker-1
这会失去自己配置的：
线程数量；
任务队列；
线程名称；
拒绝策略；
业务隔离。
正确写法是：
CompletableFuture.supplyAsync(
() -> processPdf(file),
pdfExecutor
);
日志中的线程名称会变成：
pdf-worker-1
这说明任务使用的是 PDF 专用线程池。
第五步：返回统一结果对象
真实项目里，不建议只返回文件路径。
可以定义：
PdfTaskResult
例如：
package com.succos.dto;
public class PdfTaskResult {
private final boolean success;
private final String fileName;
private final String targetPath;
private final String message;
private PdfTaskResult(
boolean success,
String fileName,
String targetPath,
String message
) {
this.success = success;
this.fileName = fileName;
this.targetPath = targetPath;
this.message = message;
}
public static PdfTaskResult success(
String fileName,
String targetPath
) {
return new PdfTaskResult(
true,
fileName,
targetPath,
"处理成功"
);
}
public static PdfTaskResult fail(
String fileName,
String message
) {
return new PdfTaskResult(
false,
fileName,
null,
message
);
}
public boolean isSuccess() {
return success;
}
public String getFileName() {
return fileName;
}
public String getTargetPath() {
return targetPath;
}
public String getMessage() {
return message;
}
}
Service 中改成：
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
return CompletableFuture
.supplyAsync(
() -> {
String targetPath =
processPdf(file);
return PdfTaskResult.success(
file.getName(),
targetPath
);
},
pdfExecutor
)
.exceptionally(ex -> {
return PdfTaskResult.fail(
file.getName(),
getErrorMessage(ex)
);
});
}
这样任务失败后也会返回一个明确结果。
增加超时控制
可以继续增加：
orTimeout(...)
完整方法如下：
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
return CompletableFuture
.supplyAsync(
() -> {
String targetPath =
processPdf(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 处理超过 30 秒"
);
}
return PdfTaskResult.fail(
file.getName(),
getErrorMessage(ex)
);
});
}
需要导入：
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
这样：
任务正常完成，返回成功结果；
任务普通异常，返回失败结果；
任务超过 30 秒，返回超时失败结果。
完整的 PdfTaskService
把前面的内容放到一起：
package com.succos.service;
import com.succos.dto.PdfTaskResult;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import java.io.File;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@Service
public class PdfTaskService {
private final ThreadPoolExecutor pdfExecutor;
public PdfTaskService(
@Qualifier("pdfExecutor")
ThreadPoolExecutor pdfExecutor
) {
this.pdfExecutor = pdfExecutor;
}
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
long start =
System.currentTimeMillis();
return CompletableFuture
.supplyAsync(
() -> {
String targetPath =
processPdf(file);
return PdfTaskResult.success(
file.getName(),
targetPath
);
},
pdfExecutor
)
.orTimeout(
30,
TimeUnit.SECONDS
)
.whenComplete((result, ex) -> {
long cost =
System.currentTimeMillis()
- start;
if (ex != null) {
System.err.println(
file.getName()
+ " 处理异常，耗时："
+ cost
+ " ms，原因："
+ getErrorMessage(ex)
);
} else {
System.out.println(
file.getName()
+ " 处理结束，耗时："
+ cost
+ " ms"
);
}
})
.exceptionally(ex -> {
Throwable cause =
getRootCause(ex);
if (cause
instanceof TimeoutException) {
return PdfTaskResult.fail(
file.getName(),
"PDF 处理超过 30 秒"
);
}
return PdfTaskResult.fail(
file.getName(),
getErrorMessage(ex)
);
});
}
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 Throwable getRootCause(
Throwable throwable
) {
Throwable cause = throwable;
while (cause.getCause() != null) {
cause = cause.getCause();
}
return cause;
}
private String getErrorMessage(
Throwable throwable
) {
Throwable cause =
getRootCause(throwable);
String message =
cause.getMessage();
if (message == null
|| message.isBlank()) {
return cause
.getClass()
.getSimpleName();
}
return message;
}
}
现在这个 Service 已经具备：
使用 Spring 线程池；
返回 CompletableFuture；
正常结果转换；
异常兜底；
超时控制；
任务耗时日志。
第六步：在 Controller 中调用 Service
新建：
com.succos.controller.PdfTaskController
为了先演示线程池调用，这里暂时使用固定文件路径：
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.io.File;
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
) {
File file =
new File(
"input",
fileName
);
return pdfTaskService
.processPdfAsync(file);
}
}
请求：
POST /pdf/watermark?fileName=test.pdf
Controller 会直接返回：
CompletableFuture<PdfTaskResult>
Spring MVC 会在异步任务完成后，把 PdfTaskResult 序列化为 JSON。
Controller 返回 CompletableFuture 会发生什么
当 Controller 返回：
CompletableFuture<PdfTaskResult>
请求线程不会在 Controller 中手动调用：
future.join();
而是把 CompletableFuture 交给 Spring MVC。
大致流程是：
HTTP 请求进入 Controller；
Controller 调用 Service；
Service 提交 PDF 异步任务；
Controller 返回 CompletableFuture；
PDF 任务在线程池中执行；
任务完成后，Spring 返回结果给客户端。
这样 Controller 不需要自己阻塞等待结果。
不要在 Controller 中立即 join
下面这种写法虽然能运行：
@PostMapping("/watermark")
public PdfTaskResult addWatermark(
@RequestParam String fileName
) {
File file =
new File("input", fileName);
return pdfTaskService
.processPdfAsync(file)
.join();
}
但这里调用了：
join()
请求线程仍然会一直等待 PDF 处理完成。
虽然 PDF 任务跑在线程池中，但接口线程被阻塞了。
如果希望使用 Spring MVC 对 CompletableFuture 的异步返回支持，可以直接返回：
CompletableFuture<PdfTaskResult>
而不是马上 join()。
直接返回结果和返回任务 ID 的区别
上面的接口虽然使用了异步线程池，但客户端仍然要等 PDF 处理完成后才能收到结果。
只是等待过程由 Spring 管理，而不是 Controller 手动 join()。
流程仍然是：
提交请求；
等待 PDF 处理完成；
返回最终结果。
如果 PDF 可能处理几十秒甚至几分钟，更合理的方式通常是：
接口接收任务；
立即返回任务 ID；
后台继续处理 PDF；
客户端根据任务 ID 查询进度。
这属于真正的后台任务系统。
后面的章节可以继续实现：
任务提交接口；
任务状态保存；
任务进度查询；
成功失败结果查询。
这一节先把线程池 Bean 和 CompletableFuture 注入方式搞清楚。
第七步：观察线程池是否被复用
连续请求三次：
POST /pdf/watermark?fileName=a.pdf
POST /pdf/watermark?fileName=b.pdf
POST /pdf/watermark?fileName=c.pdf
日志可能是：
pdf-worker-1 开始处理：a.pdf
pdf-worker-2 开始处理：b.pdf
pdf-worker-3 开始处理：c.pdf
再次请求：
POST /pdf/watermark?fileName=d.pdf
可能看到：
pdf-worker-1 开始处理：d.pdf
线程名称再次出现 pdf-worker-1，说明线程被复用了。
系统不是每次请求都创建一个新线程池或新线程。
这正是线程池存在的意义。
队列满了会发生什么
当前配置：
new ThreadPoolExecutor(
3,
3,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory("pdf-worker"),
new ThreadPoolExecutor.AbortPolicy()
)
表示：
最多 3 个任务同时执行；
最多 100 个任务排队；
第 104 个未完成任务提交时，可能被拒绝。
任务被拒绝时会抛出：
RejectedExecutionException
如果不处理，这个异常可能直接从：
CompletableFuture.supplyAsync(...)
的调用位置抛出来。
因为任务甚至没有成功提交到线程池中。
处理任务提交被拒绝
可以在调用 supplyAsync() 时增加捕获：
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
try {
return CompletableFuture
.supplyAsync(
() -> {
String targetPath =
processPdf(file);
return PdfTaskResult.success(
file.getName(),
targetPath
);
},
pdfExecutor
)
.orTimeout(
30,
TimeUnit.SECONDS
)
.exceptionally(ex -> {
return PdfTaskResult.fail(
file.getName(),
getErrorMessage(ex)
);
});
} catch (RejectedExecutionException e) {
PdfTaskResult result =
PdfTaskResult.fail(
file.getName(),
"PDF 任务过多，请稍后重试"
);
return CompletableFuture
.completedFuture(result);
}
}
需要导入：
import java.util.concurrent.RejectedExecutionException;
这里要注意：
线程池拒绝发生在任务提交阶段；
不一定进入 CompletableFuture 的 exceptionally。
所以可以在 supplyAsync() 外层单独捕获。
completedFuture 是什么
这里使用了：
CompletableFuture.completedFuture(result)
它会创建一个已经完成的 CompletableFuture。
例如：
PdfTaskResult result =
PdfTaskResult.fail(
file.getName(),
"任务提交失败"
);
return CompletableFuture
.completedFuture(result);
调用方仍然能收到：
CompletableFuture<PdfTaskResult>
只是这个 Future 不需要等待，里面已经有结果了。
这样方法的返回类型保持一致。
加入拒绝处理后的完整方法
public CompletableFuture<PdfTaskResult>
processPdfAsync(File file) {
long start =
System.currentTimeMillis();
try {
return CompletableFuture
.supplyAsync(
() -> {
String targetPath =
processPdf(file);
return PdfTaskResult.success(
file.getName(),
targetPath
);
},
pdfExecutor
)
.orTimeout(
30,
TimeUnit.SECONDS
)
.whenComplete((result, ex) -> {
long cost =
System.currentTimeMillis()
- start;
if (ex != null) {
System.err.println(
file.getName()
+ " 执行异常，耗时："
+ cost
+ " ms"
);
} else {
System.out.println(
file.getName()
+ " 执行完成，耗时："
+ cost
+ " ms"
);
}
})
.exceptionally(ex -> {
Throwable cause =
getRootCause(ex);
if (cause
instanceof TimeoutException) {
return PdfTaskResult.fail(
file.getName(),
"PDF 处理超过 30 秒"
);
}
return PdfTaskResult.fail(
file.getName(),
getErrorMessage(ex)
);
});
} catch (RejectedExecutionException e) {
System.err.println(
"PDF 任务提交被拒绝，file="
+ file.getName()
);
return CompletableFuture
.completedFuture(
PdfTaskResult.fail(
file.getName(),
"当前 PDF 任务过多，请稍后重试"
)
);
}
}
这样任务可能出现的主要状态都有了处理：
任务成功；
任务执行失败；
任务执行超时；
线程池拒绝任务。
能不能把 Bean 类型写成 Executor
可以。
例如配置类：
@Bean(name = "pdfExecutor")
public Executor pdfExecutor() {
return new ThreadPoolExecutor(
3,
3,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(100),
new NamedThreadFactory("pdf-worker"),
new ThreadPoolExecutor.AbortPolicy()
);
}
Service 中：
private final Executor pdfExecutor;
public PdfTaskService(
@Qualifier("pdfExecutor")
Executor pdfExecutor
) {
this.pdfExecutor = pdfExecutor;
}
CompletableFuture.supplyAsync() 接收的就是：
Executor
所以这样可以正常使用。
好处是业务类只依赖更抽象的接口：
Executor
不直接依赖具体的：
ThreadPoolExecutor
什么时候注入 ThreadPoolExecutor
如果业务类只负责提交任务，注入：
Executor
通常已经够了。
如果业务类还需要读取线程池状态：
pdfExecutor.getActiveCount();
pdfExecutor.getQueue().size();
pdfExecutor.getCompletedTaskCount();
那就需要注入：
ThreadPoolExecutor
不过线程池监控最好单独放到监控类中，不要让普通业务 Service 承担太多线程池管理职责。
这一阶段为了学习线程池的具体参数，可以继续使用：
ThreadPoolExecutor
Spring 容器中 Bean 的生命周期
这一版线程池的生命周期大致是：
1. Spring Boot 应用启动；
2. Spring 扫描 ThreadPoolConfig；
3. 调用 pdfExecutor()；
4. 创建 ThreadPoolExecutor；
5. 保存为名称为 pdfExecutor 的 Bean；
6. 创建 PdfTaskService；
7. 把 pdfExecutor 注入 PdfTaskService；
8. 业务运行期间不断复用线程池；
9. Spring Boot 应用关闭；
10. Spring 调用 pdfExecutor.shutdown()。
整个过程由 Spring 管理。
业务代码只需要提交任务。
常见错误一：手动调用配置类方法
配置类是：
@Configuration
public class ThreadPoolConfig {
@Bean
public ThreadPoolExecutor pdfExecutor() {
return new ThreadPoolExecutor(...);
}
}
不要在业务类里自己写：
ThreadPoolConfig config =
new ThreadPoolConfig();
ThreadPoolExecutor executor =
config.pdfExecutor();
因为这里的：
new ThreadPoolConfig()
不是由 Spring 管理的配置类对象。
直接调用方法可能重新创建一个线程池，绕过 Spring 容器。
正确方式是注入 Bean：
public PdfTaskService(
@Qualifier("pdfExecutor")
ThreadPoolExecutor pdfExecutor
)
不要自己创建配置类。
常见错误二：直接 new Service
同样不要写：
PdfTaskService service =
new PdfTaskService(...);
在 Controller 中应该通过构造方法注入：
public PdfTaskController(
PdfTaskService pdfTaskService
) {
this.pdfTaskService =
pdfTaskService;
}
这样 PdfTaskService 和它依赖的线程池都由 Spring 管理。
常见错误三：Bean 名称写错
配置类中是：
@Bean(name = "pdfExecutor")
注入时却写成：
@Qualifier("pdfThreadPool")
Spring 就找不到对应 Bean。
名称必须一致：
@Qualifier("pdfExecutor")
为了减少字符串写错，也可以定义常量：
public final class ExecutorNames {
public static final String PDF_EXECUTOR =
"pdfExecutor";
public static final String UPLOAD_EXECUTOR =
"uploadExecutor";
private ExecutorNames() {
}
}
然后：
@Bean(name = ExecutorNames.PDF_EXECUTOR)
以及：
@Qualifier(ExecutorNames.PDF_EXECUTOR)
不过小项目中直接使用明确名称也可以。
常见错误四：把线程池配置得过大
比如：
new ThreadPoolExecutor(
100,
500,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(10000),
...
)
不代表系统性能一定更好。
PDF 任务本身可能消耗：
CPU；
内存；
磁盘 IO；
文件句柄。
线程太多可能导致：
CPU 上下文切换增加；
内存占用升高；
磁盘读写争用；
应用整体响应变慢；
大量任务同时失败。
线程池大小应该结合任务性质和服务器资源确定。
学习阶段使用：
核心线程数 3；
最大线程数 3；
队列容量 100。
便于观察执行过程。
常见错误五：在所有异步步骤中忘记指定线程池
例如：
CompletableFuture
.supplyAsync(
() -> processPdf(file),
pdfExecutor
)
.thenApplyAsync(path -> {
return upload(path);
});
第一步使用：
pdfExecutor
第二步的 thenApplyAsync() 没有指定线程池，会使用默认公共线程池。
如果上传任务应该使用上传线程池，应该写成：
CompletableFuture
.supplyAsync(
() -> processPdf(file),
pdfExecutor
)
.thenApplyAsync(
path -> upload(path),
uploadExecutor
);
或者上传方法本身返回 CompletableFuture：
.thenCompose(
path -> uploadAsync(
path,
uploadExecutor
)
)
不能认为整条链会自动一直使用第一个线程池。
最终项目结构
这一节完成后，项目大致是：
src/main/java/com/succos
├── PdfApplication.java
├── config
│ ├── NamedThreadFactory.java
│ └── ThreadPoolConfig.java
├── controller
│ └── PdfTaskController.java
├── dto
│ └── PdfTaskResult.java
└── service
└── PdfTaskService.java
职责划分如下。
ThreadPoolConfig
创建线程池；
配置线程数；
配置任务队列；
配置线程名称；
配置拒绝策略；
配置销毁方法。
PdfTaskService
提交 PDF 异步任务；
执行 PDF 业务逻辑；
处理异常和超时；
返回任务结果。
PdfTaskController
接收 HTTP 请求；
调用 PdfTaskService；
把异步结果交给 Spring MVC 返回。
PdfTaskResult
统一表示任务成功或失败状态。
这一节小结
这一节我主要记住几点：
1. Spring Boot 中可以通过 @Configuration 和 @Bean 管理线程池；
2. 线程池 Bean 在应用启动时创建，在业务运行期间重复使用；
3. @Qualifier 用来从多个同类型 Bean 中选择指定线程池；
4. Service 应通过构造方法注入线程池；
5. CompletableFuture 应显式使用 Spring 管理的业务线程池；
6. 公共线程池不能在单次业务方法中调用 shutdown；
7. destroyMethod = "shutdown" 可以让 Spring 在应用关闭时销毁线程池；
8. 线程池拒绝任务时，要处理 RejectedExecutionException；
9. Controller 可以直接返回 CompletableFuture，让 Spring MVC 处理异步结果。
用一句话总结：
配置类负责创建线程池，Spring 负责管理生命周期，Service 只负责使用线程池执行任务。
下一节继续学习 Spring 提供的：
ThreadPoolTaskExecutor
它本质上还是对线程池的封装，但和 Spring 的结合更自然，并且经常配合：
@Async
实现异步方法调用。
