getResult 方法，无需用户关注如何获取到的依赖结果，只需要按照使用方式便可轻松获取所需结果值
```java
@Override
public Object task(Object o, TaskSupport support) {
// 拿到 AService (任务bean) 返回的结果， 结果类型为String类型
String result = getResult(support, AService.class, String.class);
return null;
}
```
getResult 有三个参数
* TaskSupport ： **Gobrs-Async** 所需要的核心参数，通过 task 方法中的参数透穿即可
* Class: 所依赖的 Bean的 Java 类型
* Class: 返回结果的类型
### 任务是否执行
**Gobrs-Async** 会根据 nessary 的返回结果，判断当前task 是否需要执行 如果返回true 则需要被执行，否则返之。
例如： 当参数为 cancel 时， 任务不执行。
```java
@Override
public boolean nessary(String params, TaskSupport support) {
// 假如参数是cancel 则不执行当前任务
if("cancel".equals(params)){
return false;
}
return true;
}
```
### 任务成功回调
如果你想在任务执行完成后做一些额外的操作。例如打印日志、发送邮件、发送MQ、记录信息等。 **Gobrs-Async** 同样也为你考虑到了。通过实现 callback 方法。会让你轻松的拿到
任务的执行结果。
```java
@Override
public void onSuccess(TaskSupport support) {
// todo 任务成功逻辑
}
```
### 任务失败回调
在任务异常时发送告警信息
```java
@Override
public void onFail(TaskSupport support) {
// todo 任务执行失败回调逻辑
}
```
## 重试任务
执行中的任务会出现异常的情况，如果使用者有对任务失败重试的需求，**Gobrs-Async** 也为你提供了支持,只需要在 需要开启重试的任务 bean
中使用 @Task(retryCount = 10) 注解，框架则自动会为你开启重试模式。 retryCount 所跟的数字为 **重试次数**
```java
@Component
@Task(retryCount = 10)
public class BService extends AsyncTaskAsyncTask, 唯一的区别就是以下三个步骤。
只需要在application.yml中配置
### application.yml 配置
```yaml
gobrs:
async:
config:
rules:
# 规则 是数组类型的 多组规则
- name: "general"
content: "AService->BService->FService->GService->HService;EService->CService;AService"
transaction: true
```
### 注解声明
在你需要进行事务的任务上进行回滚注解， **Gobrs-Async** 会找到 AService 任务链之前的所有任务 回调rollback 方法
```java
@Task(callback = true)
public class AService extend AsyncTask@Task注解中的 timeoutInMilliseconds 属性进行配置。
* timeoutInMilliseconds 固定使用毫秒数
```java
package com.gobrs.async.test.task.timeout;
import com.gobrs.async.core.TaskSupport;
import com.gobrs.async.core.anno.Task;
import com.gobrs.async.core.task.AsyncTask;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
/**
* The type A service.
* @program: gobrs -async-starter
* @description:
* @author: sizegang
*/
@Slf4j
@Task(timeoutInMilliseconds = 300)
public class CaseTimeoutTaskA extends AsyncTask {
/**
* The .
*/
int i = 10000;
@Override
public void prepare(Object o) {
log.info(this.getName() + " 使用线程---" + Thread.currentThread().getName());
}
@SneakyThrows
@Override
public String task(Object o, TaskSupport support) {
System.out.println("CaseTimeoutTaskA Begin");
Thread.sleep(400);
for (int i1 = 0; i1 < i; i1++) {
i1 += i1;
}
System.out.println("CaseTimeoutTaskA Finish");
return "result";
}
@Override
public boolean necessary(Object o, TaskSupport support) {
return true;
}
@Override
public void onSuccess(TaskSupport support) {
}
}
```
### 超时监听线程池配置
**所谓的监听线程池配置即：监听任务超时的线程池的核心线程数，关于为什么要配置该参数。请查看下方的[原理分析](#特别说明)。**
```yaml
gobrs:
async:
config:
rules:
- name: "chain1"
content: "taskA->taskB->taskC"
timeout-core-size: 200 # 核心线程数量
```
**如果不配置则取默认值，默认值计算方式如下：**
```java
public static Integer calculateCoreNum() {
int cpuCoreNum = Runtime.getRuntime().availableProcessors();
return new BigDecimal(cpuCoreNum).divide(new BigDecimal("0.2")).intValue();
}
```
### 单例测试
[运行代码](https://gitee.com/dromara/gobrs-async/blob/master/gobrs-async-test/src/test/java/com/gobrs/async/test/timeout/CaseTimeout.java)
### 运行结果
```sh
2022-12-09 16:51:41.031 INFO 37083 --- [pool-2-thread-1] c.g.a.t.task.timeout.CaseTimeoutTaskA : caseTimeoutTaskA 使用线程---pool-2-thread-1
CaseTimeoutTaskA Begin
2022-12-09 16:51:41.355 ERROR 37083 --- [ GobrsTimer-1] com.gobrs.async.core.timer.GobrsTimer :
com.gobrs.async.core.common.exception.TimeoutException: task caseTimeoutTaskA TimeoutException
at com.gobrs.async.core.TaskLoader$1.tick(TaskLoader.java:394) ~[classes/:na]
at com.gobrs.async.core.timer.GobrsTimer.lambda$addTimerListener$0(GobrsTimer.java:80) ~[classes/:na]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_251]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) ~[na:1.8.0_251]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) ~[na:1.8.0_251]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) ~[na:1.8.0_251]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_251]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_251]
at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_251]
2022-12-09 16:51:41.375 INFO 37083 --- [pool-2-thread-2] c.g.a.t.task.timeout.CaseTimeoutTaskB : caseTimeoutTaskB 使用线程---pool-2-thread-2
CaseTimeoutTaskB Begin
CaseTimeoutTaskB Finish
2022-12-09 16:51:41.377 INFO 37083 --- [pool-2-thread-2] c.g.a.t.task.timeout.CaseTimeoutTaskC : caseTimeoutTaskC 使用线程---pool-2-thread-2
CaseTimeoutTaskC Begin
CaseTimeoutTaskC Finish
2022-12-09 16:51:43.426 INFO 37083 --- [ main] com.gobrs.async.core.TaskLoader : 【ProcessTrace】Total cost: 2413ms【task】caseTimeoutTaskA cost :312ms【state】：fail【errMsg】: sleep interrupted; ->【task】caseTimeoutTaskB cost :0ms【state】：success; ->【task】caseTimeoutTaskC cost :2011ms【state】：success;
```
### 特别说明
* 超时任务不支持线程复用，因为需要通过控制线程超时来进行逻辑判断，如果支持线程复用，可能会出现中断正在复用线程的任务执行。
* 熔断降级原理借鉴`Hystrix` 方式处理，如果对`Hystrix`不熟悉可以借鉴 [Hystrix 熔断降级原理解析](https://my.oschina.net/13426421702/blog/3071368)
* 因线程调度的原因，超时时间可能存在10ms内的误差，可忽略！
## 线程复用
试想下方任务流程中，在理想情况下最少用几个线程即可完成流程的执行？
`taskA->taskB,TaskC->taskD->taskE,taskF` 是6个线程 还是5、4、3、2 你可能已经想到了答案是 2个线程(main线程除外)即可完成该复杂的多线程流程。 为什么呢？ 因为在并发执行时， TaskB、TaskC 又或者是
TaskE、TaskF 都只有两个并发的任务同时存在，所以决定使用线程数量个数的根本条件是有多少个**并发任务同时执行** 那么看下gobrs-async 此时有多少个线程在执行
### 测试用例
[地址](https://gitee.com/dromara/gobrs-async/blob/master/gobrs-async-test/src/test/java/com/gobrs/async/test/performance/CasePerformance.java)
### 运行结果
```sh
主线程使用main
使用main
TaskA
2022-12-11 13:58:22.581 INFO 8039 --- [ main] com.gobrs.async.core.task.AsyncTask : <11780902254572224> [taskA] execution
使用main
TaskC
使用pool-2-thread-1
TaskB
2022-12-11 13:58:22.694 INFO 8039 --- [ main] com.gobrs.async.core.task.AsyncTask : <11780902254572224> [TaskC] execution
2022-12-11 13:58:22.694 INFO 8039 --- [pool-2-thread-1] com.gobrs.async.core.task.AsyncTask : <11780902254572224> [taskB] execution
使用pool-2-thread-1
TaskD
2022-12-11 13:58:22.804 INFO 8039 --- [pool-2-thread-1] com.gobrs.async.core.task.AsyncTask : <11780902254572224> [taskD] execution
使用pool-2-thread-1
使用pool-2-thread-2
TaskE
2022-12-11 13:58:22.909 INFO 8039 --- [pool-2-thread-2] com.gobrs.async.core.task.AsyncTask : <11780902254572224> [taskE] execution
TaskF
2022-12-11 13:58:22.910 INFO 8039 --- [pool-2-thread-1] com.gobrs.async.core.task.AsyncTask : <11780902254572224> [taskF] execution
2022-12-11 13:58:22.913 INFO 8039 --- [ main] com.gobrs.async.core.TaskLoader : 【ProcessTrace】Total cost: 440ms | traceId = 11780902254572224 | 【task】taskA cost :103ms【state】：success; ->【task】taskB cost :103ms【state】：success; ->【task】TaskC cost :103ms【state】：success; ->【task】taskD cost :106ms【state】：success; ->【task】taskE cost :101ms【state】：success; ->【task】taskF cost :101ms【state】：success;
耗时462
```
### 说明
通过日志可以看到 TaskC使用了 TaskA的线程执行任务， 因TaskB 和 TaskC是并行的， 所以此时需要开辟新线程执行TaskB，等到TaskB执行完成后， TaskD继续使用
TaskB的 线程 pool-2-thread-1 执行任务， 此时TaskC执行完成后 发现其子任务已经被 TaskB释放后的线程拿到执行权，则不需要使用自身线程执行任务。 同理任务流程
继续往下执行。 整个流程中一共使用 3个线程（包含main线程）。
**小编对比业界多种多线程并发框架，目前只有gobrs-Async 具备线程复用的能力。 具体测试用例在下方性能测试章节展示。**
## 日志体系
使用 `log.error(getFormattedTraceId(), exception);` 打印的日志会自动带上 traceId， `getFormattedTraceId` 是 `AsyncTask` 提供的通用方法。
```java
@Slf4j
@Task(failSubExec = true)
public class BService extends AsyncTask {
int i = 10000;
@Override
public void prepare(Object o) {
log.info(this.getName() + " 使用线程---" + Thread.currentThread().getName());
}
@Override
public Object task(Object o, TaskSupport support) {
System.out.println("BService Begin");
for (int i1 = 0; i1 < i; i1++) {
i1 += i1;
}
try {
System.out.println(1 / 0);
} catch (Exception exception) {
log.error(getFormattedTraceId(), exception);
}
System.out.println("BService Finish");
return null;
}
}
```
## 开启全局日志打印
* `costLogabled`：打印任务流程的全链路执行过程
* `errLogabled`：开启任务异常打印 默认true
```yaml
gobrs:
async:
config:
rules:
- name: "optionalRule"
content: "caseOptionalTaskA->caseOptionalTaskB->caseOptionalTaskC,caseOptionalTaskD->caseOptionalTaskE->caseOptionalTaskF"
task-interrupt: false # 局部异常是否打断主流程 默认false
transaction: false
logConfig:
costLogabled: true # 开启任务耗时打印 log日志级别需要为 error 默认true
errLogabled: true # 开启任务异常打印 默认true
```
## 打印效果
```sh
CaseOptionalTaskA 任务执行
CaseOptionalTaskA 任务执行完成
2022-12-11 15:47:32.511 INFO 13458 --- [o-8888-exec-152] com.gobrs.async.core.task.AsyncTask : <11781331511388032> [caseOptionalTaskA] execution
CaseOptionalTaskB 任务执行
CaseOptionalTaskB 任务执行完成
2022-12-11 15:47:32.613 INFO 13458 --- [o-8888-exec-152] com.gobrs.async.core.task.AsyncTask : <11781331511388032> [caseOptionalTaskB] execution
CaseOptionalTaskD 任务执行
CaseOptionalTaskD任务执行完成
2022-12-11 15:47:32.718 INFO 13458 --- [o-8888-exec-152] com.gobrs.async.core.task.AsyncTask : <11781331511388032> [caseOptionalTaskD] execution
2022-12-11 15:47:32.718 INFO 13458 --- [o-8888-exec-152] com.gobrs.async.core.TaskLoader : 【ProcessTrace】Total cost: 311ms | traceId = 11781331511388032 | 【task】caseOptionalTaskA cost :102ms【state】：success; ->【task】caseOptionalTaskB cost :102ms【state】：success; ->【task】caseOptionalTaskD cost :105ms【state】：success;
cost 311
```
## 第三方日志框架集成
这里选择[Tlog](https://tlog.yomahub.com/) 日志框架， 使用方法这里就不做赘述了。感兴趣的小伙伴可以到官网查看
### 集成使用
```java
/**
* 启动类
*/
@SpringBootApplication
/**
* 使用gobrs-async-test 模块创建的任务 为了方便不重复创建任务了
*/
@ComponentScan(value = {"com.gobrs.async"})
public class GobrsAsyncExampleApplication {
/**
* Tlog 日志打印框架 官网: https://tlog.yomahub.com/
*/
static {
AspectLogEnhance.enhance();
}
/**
* The entry point of application.
*
* @param args the input arguments
*/
public static void main(String[] args) {
SpringApplication.run(GobrsAsyncExampleApplication.class, args);
}
}
```
### 更多细节
集成方式请参考源码示例项目 [gobrs-async-example](https://gitee.com/dromara/gobrs-async/tree/master/gobrs-async-example) 或者直接下载源码查看。
[源码地址](https://gitee.com/dromara/gobrs-async)
## 规则热更新
**在Gobrs-Async**中默认规则只会加载一次，可能有朋友可能会有规则动态变化的需求，使用程序动态修改规则配置。而不需要重新启动程序。那么对于这种需求
**Gobrs-Async** 同样支持。
> Gobrs-Async 默认使用CopyOnWrite 机制更新的规则配置，并发度更高。同时维护了线程安全机制。
```java
// 规则热加载器
@Resource
private RuleThermalLoad ruleThermalLoad;
// 热更新规则任务 无需启动程序， 只需要将规则交给 规则热加载器 即可完成接入
public void updateRule(Rule rule) {
// 单任务修改
Rule r = new Rule();
r.setName("ruleName");
r.setContent("AService->CService->EService->GService; BService->DService->FService->HService;");
ruleThermalLoad.load(rule);
// 批量修改
ListAsyncResult 的结果，判断不同的业务逻辑
**Gobrs-Async** 也为你提供了途径。 用户只需要在 异步任务的 task方法中 手动跳用 stopAsync 透传TaskSupport
无需关注内部实现，即可轻松完成关闭主流程。
stopAsync 方法有两个参数说明如下：
* TaskSupport: **Gobrs-Async** 使用参数， 透传即可。
* expCode : 中断状态码，可自定义枚举
```java
@Override
public Map task(DataContext dataContext, TaskSupport support) {
try {
// todo 执行任务
} catch (Exception e) {
// todo 根据不同的异常 处理返回不同的 中断码
if (e instanceof RedirectSimpleException) {
// 中断任务流程
stopAsync(support, PRODUCT_NOT_FOUND);
return null;
}
stopAsync(support, SYSTEM_DEMOTION);
}
return null;
}
```
[任务触发器执行完成](/pages/2f674a/#启动任务流程) 执行完成之后，会获取 AsyncResult 任务流程执行结果。
根据不同的中断码，执行不同的业务逻辑。
```java
MapTask注解中声明 failSubExec 即可继续执行任务流程。
默认 failSubExec=false
```java
@Service
@Task(failSubExec = true)
public class BService extends AsyncTask* 简化配置 *
* A,B,C->D:anyCondition *
* D根据 A,B,C 返回的任务结果中的 AnyCondition 的state状态 进行判断是否继续执行 子任务。
* @author: sizegang
* @create: 2022 -03-20
*/
@Task(failSubExec = true)
public class AServiceCondition extends AsyncTask {
/**
* The .
*/
int sums = 10000;
@Override
public AnyConditionResultapplication.yml
```yaml
gobrs:
async:
config:
rules:
# 规则 是数组类型的 多组规则
- name: "general"
content: "AService->BService->FService->GService->HService;EService->CService;AService"
task-interrupt: true # 局部异常是否打断主流程 默认 false
```
## 异常拦截器
实现 AsyncExceptionInterceptor 接口，开发一个自定义的异常处理拦截器
```java
/**
* @program: gobrs-async
* @ClassName GobrsExceptionInter
* @description: 主流程中断异常自定义处理
* @author: sizegang
* @create: 2022-02-19 22:55
* @Version 1.0
**/
@Component
public class GobrsExceptionInter implements AsyncTaskExceptionInterceptor {
@Override
public CompletionException exception(Throwable throwable, Boolean state) {
System.out.println("自定义全局异常 exceptor Interceptor 触发");
return new CompletionException(throwable);
}
}
```
## 默认情况
默认 **Gobrs-Async** 对全局拦截器开关是关闭的，如果流程中某一任务异常，只会停止所依赖该异常任务的任务停止，并调用callback 通知 下游任务
## 可配置的全局任务拦截器
有些小伙伴可能不满足与单任务的拦截，希望有一个统一拦截的入口，而不是对每一个任务做单独的处理。那么 **Gobrs-Async** 也为您提供了支持。
## 前置全局任务流程拦截器
实现 AsyncTaskPreInterceptor 接口，开发一个自定义的任务流程前置 全局拦截器。
```java
/**
* @program: m-detail
* @ClassName AsyncTaskPreInterceptor
* @description:
* @author: sizegang
* @create: 2022-03-24
**/
@Component
public class TaskPreInterceptor implements AsyncTaskPreInterceptorTaskPostInterceptor 接口，开发一个自定义的任务流程前置 全局拦截器
```java
/**
* @program: m-detail
* @ClassName AsyncTaskPreInterceptor
* @description:
* @author: sizegang
* @create: 2022-03-24
**/
@Component
public class TaskPostInterceptor implements AsyncTaskPostInterceptor {
/**
*
* @param result 任务结果
* @param taskName 任务名称
*/
@Override
public void postProcess(Object result, String taskName) {
}
}
```
## 自定义固定线程池
**Gobrs-Async** 默认使用的是 Executors.newCachedThreadPool() 的线程池， 如果你想自定义线程池。满足自己的线程池需求。
只需要 GobrsAsyncThreadPoolFactory 对象，如下：
```java
@Configuration
public class ThreadPoolConfig {
@Autowired
private GobrsAsyncThreadPoolFactory factory;
@PostConstruct
public void gobrsThreadPoolExecutor(){
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(300, 500, 30, TimeUnit.SECONDS,
new LinkedBlockingQueue());
factory.setThreadPoolExecutor(threadPoolExecutor);
}
```
## 实时更新线程池
开发者可能有这种苦恼，线程池在运行时是在项目初始化的时候从application.yml中 加载的， 一旦程序运行起来之后，就无法修改使用的线程池了。
如果自己公司有分布式配置中心，可以实时更新程序内存的应用的话，那么**gobrs**也为你提供了入口。
在我们公司是有自己的热更新组件的，所有可以如下使用：
### 配置中心的线程池配置
```json
{
corePoolSize: 210,
maxPoolSize: 600,
keepAliveTime: 30,
capacity: 10000,
threadNamePrefix: "m-detail"
rejectedExecutionHandler： "CallerRunsPolicy"
}
```
```java
@Slf4j
@Configuration
public class ThreadPoolConfig {
@Autowired
private GobrsAsyncThreadPoolFactory factory;
@Resource
private DUCCConfigService duccConfigService;
@PostConstruct
public void gobrsThreadPoolExecutor() {
// 从配置中心拿到 线程池配置规则 DuccConstant.GOBRS_ASYNC_THREAD_POOL 为线程池配置在配置中心的key
String config = duccConfigService.getString(DuccConstant.GOBRS_ASYNC_THREAD_POOL);
ThreadPool threadPool = JSONObject.parseObject(config, ThreadPool.class);
// 通过gobrs-async 提供的构造器进行构造线程池
ThreadPoolExecutor executor = ThreadPoolBuilder.buildByThreadPool(threadPool);
factory.setThreadPoolExecutor(executor);
listenerDucc();
}
// 监听配置中心 线程池改动
private void listenerDucc() {
duccConfigService.addListener(new DuccListener(DuccConstant.GOBRS_ASYNC_THREAD_POOL, property -> {
log.warn("监听到DUCC配置GobrsAsync 线程池配置变更，property：{}", JSON.toJSONString(property.getValue()));
ThreadPool threadPool = JSONObject.parseObject(property.getValue().toString(), ThreadPool.class);
ThreadPoolExecutor executor = ThreadPoolBuilder.buildByThreadPool(threadPool);
factory.setThreadPoolExecutor(executor);
// 线程池更新成功
log.warn("GobrsAsync thread pool update success");
}));
}
}
```
## 可选的执行流程
### 任务配置
```yaml
- name: "optionalRule"
content: "caseOptionalTaskA->caseOptionalTaskB->caseOptionalTaskC,caseOptionalTaskD->caseOptionalTaskE->caseOptionalTaskF"
```
如果开发者在调用时只希望执行 `caseOptionalTaskD` , 则在任务链中 只需要执行`caseOptionalTaskA、caseOptionalTaskB、caseOptionalTaskD` 三个任务即可即可。 其他任务不需要执行
，提供了随机选择流程中任务执行的能力。
### 调用方式
`gobrsAsync.go`方法的第三个参数 `Set
## 贡献者
