Spring集成asyncTool：实现复杂任务的优雅编排与高效执行
目录
一、集成到 Spring Boot
1. 添加依赖
2. 配置线程池
二、核心方法说明
1. IWorker 接口
2. ICallback 接口
3. WorkerWrapper 类
三、详细使用方式及示例
1. 串行任务
2. 并行任务
3. 阻塞等待 - 先串行，后并行
4. 阻塞等待 - 先并行，后串行
四、主要作用说明
1. 任务编排
2. 执行监控与回调
3. 异常处理与容错
4. 性能优化
5. 结果管理
6. 线程池管理
7. 简化开发
五、注意事项
一、集成到 Spring Boot
1. 添加依赖
在项目的 pom.xml 文件中添加 asyncTool 的依赖：
<dependency>
<groupId>com.jd.platform</groupId>
<artifactId>asyncTool</artifactId>
<version>版本号</version>
</dependency>2. 配置线程池
虽然 asyncTool 内部会管理线程池，但为了更好地控制线程的使用，可以自定义线程池。以下是两种配置方式：
（1）自定义线程池
@Configuration
@EnableAsync // 开启线程池
public class TaskExecutePool {
@Autowired
private TaskThreadPoolConfig config;
@Bean
public Executor myTaskAsyncPool() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(config.getCorePoolSize()); // 核心线程池大小
executor.setMaxPoolSize(config.getMaxPoolSize()); // 最大线程数
executor.setQueueCapacity(config.getQueueCapacity()); // 队列容量
executor.setKeepAliveSeconds(config.getKeepAliveSeconds()); // 活跃时间
executor.setThreadNamePrefix("MyExecutor-"); // 线程名字前缀
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 拒绝策略
executor.initialize();
return executor;
}
}（2）修改原生 Spring 异步线程池的装配
@Configuration
@EnableAsync // 开启线程池
public class NativeAsyncTaskExecutePool implements AsyncConfigurer {
@Autowired
private TaskThreadPoolConfig config;
@Bean
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(config.getCorePoolSize());
executor.setMaxPoolSize(config.getMaxPoolSize());
executor.setQueueCapacity(config.getQueueCapacity());
executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
executor.setThreadNamePrefix("MyExecutor2-");
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return (ex, method, objects) -> {
log.error("==========================" + ex.getMessage() + "=======================", ex);
log.error("exception method:" + method.getName());
};
}
}二、核心方法说明
1. IWorker 接口
-
action(T object, Map<String, WorkerWrapper> allWrappers)：任务的具体执行逻辑。
object是任务的输入参数，allWrappers是所有任务的包装类集合，可用于获取其他任务的结果。 -
defaultValue()：任务超时或异常时的默认返回值。
2. ICallback 接口
-
begin()：任务开始时的回调。
-
result(boolean success, T param, WorkResult<V> workResult)：任务执行结果的回调。
success表示任务是否成功，param是任务的输入参数，workResult是任务的执行结果。
3. WorkerWrapper 类
-
id：任务的唯一标识。
-
param：任务的输入参数。
-
worker：任务的具体实现。
-
callback：任务的回调实现。
-
depend：任务的依赖关系，定义任务的执行顺序。
-
next：任务的后续任务，用于定义任务的执行顺序。
三、详细使用方式及示例
1. 串行任务
任务按顺序依次执行。以下是一个串行任务的示例：
// 定义任务 A
WorkerWrapper wrapperA = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerA")
.worker(new WorkerA())
.callback(new WorkerA())
.param(1)
.build();
// 定义任务 B，依赖于任务 A
WorkerWrapper wrapperB = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerB")
.worker(new WorkerB())
.callback(new WorkerB())
.param(2)
.depend(wrapperA)
.build();
// 定义任务 C，依赖于任务 B
WorkerWrapper wrapperC = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerC")
.worker(new WorkerC())
.callback(new WorkerC())
.param(3)
.depend(wrapperB)
.build();
// 提交任务
Async.beginWork(1000, wrapperA);2. 并行任务
多个任务同时执行。以下是一个并行任务的示例：
// 定义任务 A
WorkerWrapper wrapperA = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerA")
.worker(new WorkerA())
.callback(new WorkerA())
.param(1)
.build();
// 定义任务 B
WorkerWrapper wrapperB = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerB")
.worker(new WorkerB())
.callback(new WorkerB())
.param(2)
.build();
// 定义任务 C
WorkerWrapper wrapperC = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerC")
.worker(new WorkerC())
.callback(new WorkerC())
.param(3)
.build();
// 提交任务
Async.beginWork(1000, wrapperA, wrapperB, wrapperC);3. 阻塞等待 - 先串行，后并行
先执行任务 A，然后任务 B 和任务 C 并行执行：
// 定义任务 A
WorkerWrapper wrapperA = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerA")
.worker(new WorkerA())
.callback(new WorkerA())
.param(1)
.build();
// 定义任务 B，依赖于任务 A
WorkerWrapper wrapperB = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerB")
.worker(new WorkerB())
.callback(new WorkerB())
.param(2)
.depend(wrapperA)
.build();
// 定义任务 C，依赖于任务 A
WorkerWrapper wrapperC = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerC")
.worker(new WorkerC())
.callback(new WorkerC())
.param(3)
.depend(wrapperA)
.build();
// 提交任务
Async.beginWork(1000, wrapperA);4. 阻塞等待 - 先并行，后串行
任务 B 和任务 C 并行执行，完成后任务 A 执行：
// 定义任务 A
WorkerWrapper wrapperA = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerA")
.worker(new WorkerA())
.callback(new WorkerA())
.param(null) // 参数为任务 B 和任务 C 的结果
.build();
// 定义任务 B
WorkerWrapper wrapperB = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerB")
.worker(new WorkerB())
.callback(new WorkerB())
.param(2)
.next(wrapperA)
.build();
// 定义任务 C
WorkerWrapper wrapperC = new WorkerWrapper.Builder<Integer, Integer>()
.id("workerC")
.worker(new WorkerC())
.callback(new WorkerC())
.param(3)
.next(wrapperA)
.build();
// 提交任务
Async.beginWork(1000, wrapperB, wrapperC);四、主要作用说明
1. 任务编排
-
灵活的并行与串行组合：
asyncTool支持任意组合多线程的并行和串行任务，开发者可以根据业务需求灵活定义任务的执行顺序。 -
任务依赖管理：它允许任务之间存在强依赖和弱依赖关系。例如，某些任务必须在其他任务完成后才能执行，而另一些任务则可以在依赖任务中的任意一个或多个完成后执行。
2. 执行监控与回调
-
全链路回调机制：每个任务在执行过程中，无论成功、失败、超时还是异常，都会触发回调函数。这使得开发者可以实时监控任务的执行状态。
-
任务跳过回调：即使某些任务被跳过未执行，
asyncTool也会提供回调，方便开发者进行日志记录或异常处理。
3. 异常处理与容错
-
异常与超时处理：每个任务可以设置超时时间和默认值，当任务执行失败或超时时，会返回默认值，确保整个任务链的稳定性。
-
独立任务容错：单个任务的失败不会影响其他任务的回调和最终结果的获取，但如果任务依赖的上游任务失败，则当前任务也会失败并返回默认值。
4. 性能优化
-
低线程设计：
asyncTool采用低线程设计，减少线程的创建和销毁开销。例如，在多个任务依赖关系中，后续任务可以复用前一个任务的线程。 -
无锁机制：整个框架全程无锁，避免了锁带来的性能开销，提高了并发性能。
5. 结果管理
-
按顺序返回结果：任务执行完成后，
asyncTool可以按任务添加的顺序返回结果列表，方便开发者进行后续处理。 -
支持异步回调：除了同步阻塞返回结果外，还支持整个任务组的异步回调，避免阻塞主线程。
6. 线程池管理
-
线程池共享与独占：支持为每个任务组独享线程池，也可以让所有任务组共享一个线程池，灵活配置资源。
7. 简化开发
-
封装复杂逻辑：
asyncTool封装了复杂的并发逻辑，使得开发者可以更专注于业务逻辑的实现，而无需深入了解底层的并发机制。
五、注意事项
-
任务的线程安全：由于任务可能在多个线程中并发执行，需要确保任务的线程安全性。
-
任务的异常处理：在任务执行过程中可能会出现异常，需要合理地处理异常，避免影响整个应用的运行。
-
任务的超时设置：合理设置任务的超时时间，避免任务长时间未完成导致资源浪费。
-
任务的依赖关系：正确配置任务的依赖关系，确保任务按预期顺序执行。
通过以上详细说明和代码示例，你可以在 Spring Boot 项目中灵活使用 asyncTool 实现复杂的多线程任务编排。