⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处网络「网络」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

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

芋道源码 —— 纯源码解析博客

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