Spring + asyncTool：实现复杂任务的编排

一、集成到 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);