使用京东AsyncTool实现异步编排

 asyncTool: 解决任意的多线程并行、串行、阻塞、依赖、回调的并行框架，可以任意组合各线程的执行顺序，带全链路执行结果回调。多线程编排一站式解决方案。来自于京东主App后台。

/**
     * 批量更新用户（使用京东AsyncTool）
     *
     * @param idList
     */
    public void batchUpdateAsyncTool(List<Long> idList) {
        Boolean resultBoolean = processOrderAsyncTool(idList);
        if (resultBoolean) {
            System.out.println("所有都执行完毕");
        }
    }

@Transactional(rollbackFor = Exception.class)
    public Boolean processOrderAsyncTool(List<Long> idList) {
        List<WorkerWrapper> workerWrapperList = new ArrayList<>();
        for (int i = 0; i < idList.size(); i++) {
            UserServiceUpdateAsyncTool userServiceUpdateAsyncToolObject = new UserServiceUpdateAsyncTool();
            WorkerWrapper<Long, Boolean> workerWrapper = new WorkerWrapper.Builder<Long, Boolean>()
                    .id("wrapper" + idList.get(i))
                    .worker(userServiceUpdateAsyncToolObject)
                    .callback(userServiceUpdateAsyncToolObject)
                    .param(idList.get(i))//1+1
                    .build();
            workerWrapperList.add(workerWrapper);
        }
        try {
            //3个WorkerWrapper一起begin
            com.jd.platform.async.executor.Async.beginWork(1000, asyncExecutor, workerWrapperList);
        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
        }
        workerWrapperList.stream().forEach(workerWrapper -> {
            System.out.println("workResult:" + workerWrapper.getWorkResult());
            if ("EXCEPTION".equals(workerWrapper.getWorkResult().getResultState().name())) {
                if (workerWrapper.getWorkResult().getEx() instanceof BusinessException) {
                    throw new BusinessException(workerWrapper.getWorkResult().getEx().getMessage());
                } else {
                    throw new RuntimeException(workerWrapper.getWorkResult().getEx().getMessage());
                }
            }
        });
        return Boolean.TRUE;
    }

@Slf4j
@Service
public class UserServiceUpdateAsyncTool implements IWorker<Long, Boolean>, ICallback<Long, Boolean> {

    private UserService userService;


    @Transactional(rollbackFor = Exception.class)
    public Boolean funcAsyncTool(Long id) {
        System.out.println("lo开始=" + id);
        userService = ApplicationContextHolder.context.getBean(UserService.class);
//        try {
//            Thread.sleep(5000);
        User user = new User();
        user.setAge(Integer.valueOf(String.valueOf(Long.valueOf("30") + id)));
        if (id.equals(2L)) {
            throw new BusinessException("出现了2异常");
        }
        else if (id.equals(3L)) {
            throw new RuntimeException("出现了3异常");
        }
        }
        this.userService.update(user, Wrappers.lambdaUpdate(User.class).eq(User::getId, id));
//        } catch (InterruptedException e) {

//        } catch (BusinessException e) {

//        } catch (RuntimeException e) {

//        }
        System.out.println("lo结束=" + id);
        return Boolean.TRUE;
    }

    @Override
    public void begin() {

    }

    @Override
    public void result(boolean b, Long s, WorkResult<Boolean> workResult) {
//        if ("EXCEPTION".equals(workResult.getResultState().name())) {
//            throw new BusinessException(workResult.getEx().getMessage());
//        }
    }

    @Override
    public Boolean action(Long s, Map<String, WorkerWrapper> map) {
        return this.funcAsyncTool(s);
    }

    @Override
    public Boolean defaultValue() {
        return null;
    }
}

运行结果：

{
    "success": false,
    "code": "505",
    "message": "出现了2异常",
    "data": null
}

入参：[1,2,3,4,5]

运行结果，2、3没有更新，更新了1、4、5