asyncTool/README.md
2019-12-26 14:07:04 +08:00

12 KiB
Raw Blame History

并发框架说明

并发场景可能存在的需求之——任意编排

1 多个执行单元的串行请求

输入图片说明

2 多个执行单元的并行请求

输入图片说明

3 阻塞等待,串行的后面跟多个并行

输入图片说明

4 阻塞等待,多个并行的执行完毕后才执行某个

输入图片说明

5 串并行相互依赖

输入图片说明

6 复杂场景

输入图片说明

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

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

此时,传统的就无能为力了。

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

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

如上图的3A和B并发执行最后是C。

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

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

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

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

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

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

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

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

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

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

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

该框架全程无锁,没有一个加锁的地方。

创建线程量少。输入图片说明 如这样的A会运行在B、C执行更慢的那个单元的线程上而不会额外创建线程。

async-Tool介绍

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

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

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

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

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

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

该框架全程无锁。

基本组件

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

TV两个泛型分别是入参和出参类型。

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

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

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

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

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

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

        workerWrapper.addNext(workerWrapper1, workerWrapper2);
        workerWrapper1.addNext(workerWrapper3);
        workerWrapper2.addNext(workerWrapper3);

输入图片说明

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无法做到的。

安装教程

代码不多,直接拷贝包过去即可。

使用说明

  1. 3个任务并行

输入图片说明

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

        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);
        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(getThreadCount());

        System.out.println(workerWrapper.getWorkResult());
//        System.out.println(getThreadCount());
        Async.shutDown();
  1. 1个执行完毕后开启另外两个另外两个执行完毕后开始第4个

输入图片说明

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

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

        workerWrapper.addNext(workerWrapper1, workerWrapper2);
        workerWrapper1.addNext(workerWrapper3);
        workerWrapper2.addNext(workerWrapper3);

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

        //正常完毕
        Async.beginWork(4100, workerWrapper);
        //3会超时
//        Async.beginWork(3100, workerWrapper);
        //2,3会超时
//        Async.beginWork(2900, workerWrapper);

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

        System.out.println(getThreadCount());
        Async.shutDown();
  1. 复杂点的

输入图片说明

做测试类里能找到下图是执行结果。看时间戳就知道执行的顺序。每个执行单元都是睡1秒。

输入图片说明

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