并发编程大杀器,京东多线程编排工具asyncTool

发布时间:2023年12月29日

图片

一、简介

并发编程大杀器,京东多线程编排工具asyncTool,可以解决任意的多线程并行、串行、阻塞、依赖、回调的并行框架,可以任意组合各线程的执行顺序,带全链路执行结果回调。多线程编排一站式解决方案。

二、特点

  • 多线程编排;

  • 全链路回调;

  • 超时控制;

解决任意的多线程并行、串行、阻塞、依赖、回调的并发框架,可以任意组合各线程的执行顺序,带全链路回调和超时控制。

其中的A、B、C分别是一个最小执行单元(worker),可以是一段耗时代码、一次Rpc调用等,不局限于你做什么。

该框架可以将这些worker,按照你想要的各种执行顺序,加以组合编排。最终得到结果。

并且,该框架 为每一个worker都提供了执行结果的回调和执行失败后自定义默认值 。譬如A执行完毕后,A的监听器会收到回调,带着A的执行结果(成功、超时、异常)。

根据你的需求,将各个执行单元组合完毕后,开始在主线程执行并阻塞,直到最后一个执行完毕。并且 可以设置全组的超时时间 。

该框架支持后面的执行单元以前面的执行单元的结果为自己的入参 。譬如你的执行单元B的入参是ResultA,ResultA就是A的执行结果,那也可以支持。在编排时,就可以预先设定B或C的入参为A的result,即便此时A尚未开始执行。当A执行完毕后,自然会把结果传递到B的入参去。

三、并行场景

1.?客户端请求服务端接口,该接口需要调用其他N个微服务的接口

譬如 请求我的购物车,那么就需要去调用用户的rpc、商品详情的rpc、库存rpc、优惠券等等好多个服务。同时,这些服务还有相互依赖关系,譬如必须先拿到商品id后,才能去库存rpc服务请求库存信息。最终全部获取完毕后,或超时了,就汇总结果,返回给客户端。

2. 并行执行N个任务,后续根据这1-N个任务的执行结果来决定是否继续执行下一个任务

如用户可以通过邮箱、手机号、用户名登录,登录接口只有一个,那么当用户发起登录请求后,我们需要并行根据邮箱、手机号、用户名来同时查数据库,只要有一个成功了,都算成功,就可以继续执行下一步。而不是先试邮箱能否成功、再试手机号……

再如某接口限制了每个批次的传参数量,每次最多查询10个商品的信息,我有45个商品需要查询,就可以分5堆并行去查询,后续就是统计这5堆的查询结果。就看你是否强制要求全部查成功,还是不管有几堆查成功都给客户做返回

再如某个接口,有5个前置任务需要处理。其中有3个是必须要执行完毕才能执行后续的,另外2个是非强制的,只要这3个执行完就可以进行下一步,到时另外2个如果成功了就有值,如果还没执行完,就是默认值。

3. 需要进行线程隔离的多批次任务

如多组任务, 各组任务之间彼此不相关,每组都需要一个独立的线程池,每组都是独立的一套执行单元的组合。有点类似于hystrix的线程池隔离策略。

4. 单机工作流任务编排

5. 其他有顺序编排的需求

四、应用需求

1. 并行场景可能存在的需求之——每个执行结果的回调

传统的Future、CompleteableFuture一定程度上可以完成任务编排,并可以把结果传递到下一个任务。如CompletableFuture有then方法,但是却无法做到对每一个执行单元的回调。譬如A执行完毕成功了,后面是B,我希望A在执行完后就有个回调结果,方便我监控当前的执行状况,或者打个日志什么的。失败了,我也可以记录个异常信息什么的。

此时,CompleteableFuture就无能为力了。

我的框架提供了这样的回调功能。并且,如果执行异常、超时,可以在定义这个执行单元时就设定默认值。

2. 并行场景可能存在的需求之——执行顺序的强依赖和弱依赖

有些场景下,我们希望B和C都执行完毕后,才能执行A,CompletableFuture里有个allOf(futures...).then()方法可以做到。

有些场景下,我们希望B或者C任何一个执行完毕,就执行A,CompletableFuture里有个anyOf(futures...).then()方法可以做到。

我的框架同样提供了类似的功能,通过设定wrapper里的addDepend依赖时,可以指定依赖的任务是否must执行完毕。如果依赖的是must要执行的,那么就一定会等待所有的must依赖项全执行完毕,才执行自己。

如果依赖的都不是must,那么就可以任意一个依赖项执行完毕,就可以执行自己了。

注意:这个依赖关系是有必须和非必须之分的,还有一个重要的东西是执行单元不能重复执行。譬如图4,如果B执行完毕,然后执行了A,此时C终于执行完了,然后也到了A,此时就会发现A已经在执行,或者已经完毕(失败),那么就不应该再重复执行A。

还有一种场景,如下图,A和D并行开始,D先执行完了,开始执行Result任务,此时B和C都还没开始,然后Result执行完了,虽然B和C都还没执行,但是已经没必要执行了。B和C这些任务是可以被跳过的,跳过的原则是他们的NextWrapper已经有结果了或者已经在执行了。我提供了checkNextWrapperResult方法来控制,当后面的任务已经执行了,自己还要不要执行的逻辑控制。当然,这个控制仅限于nextWrapper只有一个时才成立。

3. 并发场景可能存在的需求之——依赖上游的执行结果作为入参

譬如A-B-C三个执行单元,A的入参是String,出参是int,B呢它需要用A的结果作为自己的入参。也就是说A、B并不是独立的,而是有结果依赖关系的。

在A执行完毕之前,B是取不到结果的,只是知道A的结果类型。

那么,我的框架也支持这样的场景。可以在编排时,就取A的结果包装类,作为B的入参。虽然此时尚未执行,必然是空,但可以保证A执行完毕后,B的入参会被赋值。

在V1.3后,框架支持在worker的action的入参Map<String, WorkerWrapper>中获取任意一个执行单元的执行结果,当然,可以取其中的1个、多个执行结果作为自己的入参。Key就是在定义wrapper时通过id传进来的唯一id标识。详情demo可以查看test包下dependnew包案例。

4. 并发场景可能存在的需求之——全组任务的超时

一组任务,虽然内部的各个执行单元的时间不可控,但是我可以控制全组的执行时间不超过某个值。通过设置timeOut,来控制全组的执行阈值。

5. 并发场景可能存在的需求之——高性能、低线程数

该框架全程无锁,不依靠线程锁来保证顺序。

五、基本组件

1. worker

一个最小的任务执行单元。通常是一个网络调用,或一段耗时操作。

T,V两个泛型,分别是入参和出参类型。

譬如该耗时操作,入参是String,执行完毕的结果是Integer,那么就可以用泛型来定义。

多个不同的worker之间,没有关联,分别可以有不同的入参、出参类型。


/**
 * 每个最小执行单元需要实现该接口
 * @author wuweifeng wrote on 2019-11-19.
 */
public interface IWorker<T, V> {
    /**
     * 在这里做耗时操作,如rpc请求、IO等
     *
     * @param object
     *         object
     */
    V action(T object, Map<String, WorkerWrapper> allWrappers);

    /**
     * 超时、异常时,返回的默认值
     * @return 默认值
     */
    V defaultValue();
}

2. callBack

对每个worker的回调。worker执行完毕后,会回调该接口,带着执行成功、失败、原始入参、和详细的结果。


/**
 * 每个执行单元执行完毕后,会回调该接口</p>
 * 需要监听执行结果的,实现该接口即可
 * @author wuweifeng wrote on 2019-11-19.
 */
public interface ICallback<T, V> {

    void begin();

    /**
     * 耗时操作执行完毕后,就给value注入值
     *
     */
    void result(boolean success, T param, WorkResult<V> workResult);
}

3. wrapper

组合了worker和callback,是一个 最小的调度单元 。通过编排wrapper之间的关系,达到组合各个worker顺序的目的。

wrapper的泛型和worker的一样,决定了入参和结果的类型。

WorkerWrapper<String, String> workerWrapper = new WorkerWrapper<>(w, "0", w);
WorkerWrapper<String, String> workerWrapper1 = new WorkerWrapper<>(w1, "1", w1);
WorkerWrapper<String, String> workerWrapper2 = new WorkerWrapper<>(w2, "2", w2);
WorkerWrapper<String, String> workerWrapper3 = new WorkerWrapper<>(w3, "3", w3);

图片

0执行完,同时1和2, 1\2都完成后3。3会等待2完成

此时,你可以定义一个 worker


/**
 * @author wuweifeng wrote on 2019-11-20.
 */
public class ParWorker1 implements IWorker<String, String>, ICallback<String, String> {

    @Override
    public String action(String object) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "result = " + SystemClock.now() + "---param = " + object + " from 1";
    }

    @Override
    public String defaultValue() {
        return "worker1--default";
    }

    @Override
    public void begin() {
        //System.out.println(Thread.currentThread().getName() + "- start --" + System.currentTimeMillis());
    }

    @Override
    public void result(boolean success, String param, WorkResult<String> workResult) {
        if (success) {
            System.out.println("callback worker1 success--" + SystemClock.now() + "----" + workResult.getResult()
                    + "-threadName:" +Thread.currentThread().getName());
        } else {
            System.err.println("callback worker1 failure--" + SystemClock.now() + "----"  + workResult.getResult()
                    + "-threadName:" +Thread.currentThread().getName());
        }
    }

}

通过这一个类看一下,action里就是你的耗时操作,begin就是任务开始执行时的回调,result就是worker执行完毕后的回调。当你组合了多个执行单元时,每一步的执行,都在掌控之内。失败了,还会有自定义的默认值。这是CompleteableFuture无法做到的。

六、项目引入

外网使用jitpack.io上打的包 先添加repositories节点

  <repositories>
    <repository>
        <id>jitpack.io</id>
        <url>https://jitpack.io</url>
    </repository>
  </repositories>

然后添加如下maven依赖

  <dependency>
      <groupId>com.gitee.jd-platform-opensource</groupId>
      <artifactId>asyncTool</artifactId>
      <version>V1.4-SNAPSHOT</version>
  </dependency>

七、使用说明

1.?三个任务并行

图片

        ParWorker w = new ParWorker();
        ParWorker1 w1 = new ParWorker1();
        ParWorker2 w2 = new ParWorker2();

        WorkerWrapper<String, String> workerWrapper2 =  new WorkerWrapper.Builder<String, String>()
                .worker(w2)
                .callback(w2)
                .param("2")
                .build();

        WorkerWrapper<String, String> workerWrapper1 =  new WorkerWrapper.Builder<String, String>()
                .worker(w1)
                .callback(w1)
                .param("1")
                .build();

        WorkerWrapper<String, String> workerWrapper =  new WorkerWrapper.Builder<String, String>()
                .worker(w)
                .callback(w)
                .param("0")
                .build();

        long now = SystemClock.now();
        System.out.println("begin-" + now);

        Async.beginWork(1500, workerWrapper, workerWrapper1, workerWrapper2);
//        Async.beginWork(800, workerWrapper, workerWrapper1, workerWrapper2);
//        Async.beginWork(1000, workerWrapper, workerWrapper1, workerWrapper2);

        System.out.println("end-" + SystemClock.now());
        System.err.println("cost-" + (SystemClock.now() - now));
        System.out.println(Async.getThreadCount());

        System.out.println(workerWrapper.getWorkResult());
        Async.shutDown();
       

2. 一个执行完毕后,开启另外两个,另外两个执行完毕后,开始第四个

图片

        ParWorker w = new ParWorker();
        ParWorker1 w1 = new ParWorker1();
        ParWorker2 w2 = new ParWorker2();
        ParWorker3 w3 = new ParWorker3();

        WorkerWrapper<String, String> workerWrapper3 =  new WorkerWrapper.Builder<String, String>()
                .worker(w3)
                .callback(w3)
                .param("3")
                .build();

        WorkerWrapper<String, String> workerWrapper2 =  new WorkerWrapper.Builder<String, String>()
                .worker(w2)
                .callback(w2)
                .param("2")
                .next(workerWrapper3)
                .build();

        WorkerWrapper<String, String> workerWrapper1 =  new WorkerWrapper.Builder<String, String>()
                .worker(w1)
                .callback(w1)
                .param("1")
                .next(workerWrapper3)
                .build();

        WorkerWrapper<String, String> workerWrapper =  new WorkerWrapper.Builder<String, String>()
                .worker(w)
                .callback(w)
                .param("0")
                .next(workerWrapper1, workerWrapper2)
                .build();


        long now = SystemClock.now();
        System.out.println("begin-" + now);

        Async.beginWork(3100, workerWrapper);
//        Async.beginWork(2100, workerWrapper);

        System.out.println("end-" + SystemClock.now());
        System.err.println("cost-" + (SystemClock.now() - now));

        System.out.println(Async.getThreadCount());
        Async.shutDown();

如果觉得这样不符合左右的顺序,也可以用这种方式:

        WorkerWrapper<String, String> workerWrapper =  new WorkerWrapper.Builder<String, String>()
                .worker(w)
                .callback(w)
                .param("0")
                .build();

        WorkerWrapper<String, String> workerWrapper3 =  new WorkerWrapper.Builder<String, String>()
                .worker(w3)
                .callback(w3)
                .param("3")
                .build();

        WorkerWrapper<String, String> workerWrapper2 =  new WorkerWrapper.Builder<String, String>()
                .worker(w2)
                .callback(w2)
                .param("2")
                .depend(workerWrapper)
                .next(workerWrapper3)
                .build();

        WorkerWrapper<String, String> workerWrapper1 =  new WorkerWrapper.Builder<String, String>()
                .worker(w1)
                .callback(w1)
                .param("1")
                .depend(workerWrapper)
                .next(workerWrapper3)
                .build();

3.?依赖别的worker执行结果作为入参

可以从action的参数中根据wrapper的id获取任意一个执行单元的执行结果,但请注意执行顺序,如果尚未执行,则在调用WorkerResult.getResult()会得到null!

图片

图片

其他的详见test包下的测试类,支持各种形式的组合、编排。

gitee地址:

https://gitee.com/jd-platform-opensource/asyncTool

图片

文章来源:https://blog.csdn.net/weixin_40381772/article/details/135289694
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。