Spring + asyncTool：实现复杂任务的优雅编排与高效执行

👉 欢迎加入小哈的星球，你将获得:专属的项目实战（多个项目） / 1v1 提问 /Java 学习路线 /学习打卡 / 每月赠书 / 社群讨论

• 新项目：《Spring AI 项目实战》正在更新中..., 基于 Spring AI + Spring Boot 3.x + JDK 21;

• 《从零手撸：仿小红书（微服务架构）》 已完结，基于 Spring Cloud Alibaba + Spring Boot 3.x + JDK 17..., 点击查看项目介绍；演示地址：http://116.62.199.48:7070/

• 《从零手撸：前后端分离博客项目（全栈开发）》2期已完结,演示链接：http://116.62.199.48/;

• 专栏阅读地址：https://www.quanxiaoha.com/column
  

截止目前，累计输出 100w+ 字，讲解图 4013+ 张，还在持续爆肝中..后续还会上新更多项目，目标是将 Java 领域典型的项目都整一波，如秒杀系统, 在线商城, IM 即时通讯，Spring Cloud Alibaba 等等，戳我加入学习，解锁全部项目，已有4200+小伙伴加入

一、集成到 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实现复杂的多线程任务编排。

👉 欢迎加入小哈的星球，你将获得:专属的项目实战（多个项目） / 1v1 提问 /Java 学习路线 /学习打卡 / 每月赠书 / 社群讨论

• 新项目：《Spring AI 项目实战》正在更新中..., 基于 Spring AI + Spring Boot 3.x + JDK 21;

• 《从零手撸：仿小红书（微服务架构）》 已完结，基于 Spring Cloud Alibaba + Spring Boot 3.x + JDK 17..., 点击查看项目介绍；演示地址：http://116.62.199.48:7070/

• 《从零手撸：前后端分离博客项目（全栈开发）》2期已完结,演示链接：http://116.62.199.48/;

• 专栏阅读地址：https://www.quanxiaoha.com/column
  

截止目前，累计输出 100w+ 字，讲解图 4013+ 张，还在持续爆肝中..后续还会上新更多项目，目标是将 Java 领域典型的项目都整一波，如秒杀系统, 在线商城, IM 即时通讯，Spring Cloud Alibaba 等等，戳我加入学习，解锁全部项目，已有4200+小伙伴加入

1. 我的私密学习小圈子，从0到1手撸企业实战项目~ 2. 这才是企业级的oss-spring-boot-starter，属实好用！ 3. 面试官：String、StringBuilder 和 StringBuffer 的区别？ 4. MyBatis 批量插入从5分钟缩短到3秒？我的三个关键优化

最近面试BAT，整理一份面试资料《Java面试BATJ通关手册》，覆盖了Java核心技术、JVM、Java并发、SSM、微服务、数据库、数据结构等等。 获取方式：点“在看”，关注公众号并回复 Java 领取，更多内容陆续奉上。
PS：因公众号平台更改了推送规则，如果不想错过内容，记得读完点一下“在看”，加个“星标”，这样每次新文章推送才会第一时间出现在你的订阅列表里。 点“在看”支持小哈呀，谢谢啦