dromara/gobrs-async

    @Override
    public Boolean task(Object o, TaskSupport support) {
        try {
            System.out.println("AServiceCondition Begin");
            Thread.sleep(300);
            for (int i1 = 0; i1 < i; i1++) {
                i1 += i1;
            }
            System.out.println("AServiceCondition Finish");
        } catch (InterruptedException e) {
           // log 打印
           //e.printStackTrace();
           // 返回false 则没有资格执行子任务的逻辑
           return false;
         
        }
        // 返回true 进行执行子任务的逻辑
        return true;
    }

@Task(name = "itemTask")

/**
 * @program: gobrs-async
 * @ClassName TaskBean
 * @description: 异步任务 任务参数 Object类型 ; 任务返回 Object类型
 **/
@Task
public class BService extends AsyncTask<Object, Object> {

    @Override
    public void prepare(Object o) {
        // 任务开始之前的操作 即task 之前
    }

    @Override
    public Object task(Object o, TaskSupport support) {
        String result = getResult(support, AService.class);
        System.out.println("拿到的结果" + result);
        System.out.println("执行BService");
        return null;
    }

    // 什么条件下触发 task执行
    @Override
    public boolean nessary(Object o, TaskSupport support) {
        return true;
    }

    // task执行成功回调用
    @Override
    public void onSuccess(TaskSupport support) {

    }
    
    // task 执行失败 回调
    @Override
    public void onFail(TaskSupport support) {

    }
}

@Override
public Object task(Object o, TaskSupport support) {
    // todo 业务逻辑
    return null;
}

@Override
public Object task(Object o, TaskSupport support) {
    // 拿到 AService (任务bean) 返回的结果， 结果类型为String类型
    String result = getResult(support, AService.class, String.class);
    
    return null;
}

@Override
public boolean nessary(String params, TaskSupport support) {
     // 假如参数是cancel 则不执行当前任务
    if("cancel".equals(params)){
        return false;
    }
    return true;
}

@Override
public void onSuccess(TaskSupport support) {
    // todo 任务成功逻辑
}

@Override
public void onFail(TaskSupport support) {
    // todo  任务执行失败回调逻辑
}

@Component
@Task(retryCount = 10)
public class BService extends AsyncTask<Object, Object>  {
// ...
}

@Task(callback = true)
public class AService extend AsyncTask<Object,Object>{
 // ...
}

//  事务回滚 具体回滚业务需要自己实现 该方法是一个默认方法 需要自己手动重写
@Override
public void rollback(DataContext dataContext) {
    //super.rollback(dataContext);
    //todo rollback business
}

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) {
    }
}

    public static Integer calculateCoreNum() {
        int cpuCoreNum = Runtime.getRuntime().availableProcessors();
        return new BigDecimal(cpuCoreNum).divide(new BigDecimal("0.2")).intValue();
    }

@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;
    }
}

/**
 * 启动类
 */
@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);
    }
}

// 规则热加载器
@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);
    
    // 批量修改 
    List<Rule> updateRules = new ArrayList<Rule>();
    updateRules.add(r);
    // updateRules.add(...);
    ruleThermalLoad.load(updateRules);
}

    @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;
    }

Map<String, Object> params  = new HashMap();
// 任务流程名称 , 任务流程传入参数, 任务流程超时时间 
AsyncResult asyncResult = gobrsAsync.go("ruleName", () -> params, timeOut);

if(asyncResult.getExpCode().equals(100)){
    // 业务一
}else if(asyncResult.getExpCode().equals(200)){
    // 业务二
}

@Service
@Task(failSubExec = true) 
public class BService extends AsyncTask<Object, Object>  {
// ...
}

package com.gobrs.async.test.task.condition;

import com.gobrs.async.core.TaskSupport;
import com.gobrs.async.core.anno.Task;
import com.gobrs.async.core.common.domain.AnyConditionResult;
import com.gobrs.async.core.task.AsyncTask;
import org.springframework.stereotype.Component;

/**
 * The type A service.
 *
 * @program: gobrs -async-starter
 * @ClassName AService
 * @description: 任务依赖类型
 * AServiceCondition,BServiceCondition,CServiceCondition->DServiceCondition:anyCondition
 * <p>
 * 简化配置
 * <p>
 * A,B,C->D:anyCondition
 * <p>
 * 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 AnyConditionResult<String> task(Object o, TaskSupport support) {
        AnyConditionResult.Builder<String> builder = AnyConditionResult.builder();
        try {
            System.out.println("AServiceCondition Begin");
            Thread.sleep(300);
            for (int i1 = 0; i1 < sums; i1++) {
                i1 += i1;
            }
            System.out.println("AServiceCondition Finish");
        } catch (InterruptedException e) {
            e.printStackTrace();
            //  异常返回false
           return builder.setState(false).build();
        }
        return builder.setState(true).build();
    }
}

/**
 * @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);
    }
}

/**
 * @program: m-detail
 * @ClassName AsyncTaskPreInterceptor
 * @description:
 * @author: sizegang
 * @create: 2022-03-24
 **/
@Component
public class TaskPreInterceptor implements AsyncTaskPreInterceptor<DataContext> {
    
     /**
     * 
     * @param params 参数
     * @param taskName 任务名称
     */
    @Override
    public void preProcess(DataContext params, String taskName) {

    }
}

/**
 * @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) {

    }
}

@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);
    }

@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");
        }));
    }

}

    @Test
    public void testOptional() {
        Map<Class, Object> params = new HashMap<>();
        Set<String> options = new HashSet<>();
        options.add("caseOptionalTaskD"); # options中添加要执行的任务 bean 名称
        AsyncResult asyncResult = gobrsAsync.go("optionalRule", () -> params, options, 300000);
    }

static {
    GobrsLogger.logger();
}