zoukankan      html  css  js  c++  java
  • Fork/Join框架原理

    Fork/Join是一个用户并行执行任务的框架,把一个大任务分割成若干个小人物,最终汇总每个小任务的结果后得到大任务的结果。

    讲到Fork/Join必须要讲工作窃取算法。

    下面讲下什么叫工作窃取算法:某个线程从其他队列里窃取任务来执行,听名字感觉很奇怪,不经想什么鬼?黑人问号脸。。。。

    我们来举个例子:假如我们需要做一个比较大的任务,可以把这个任务分割为若干互不相依的子任务,为了减少线程间的竞争,把这些子任务放到不同的队列里,并为每个队列创建一个单独的线程来执行队列里的任务,线程和队列对应。这时A线程负责的A队列里的任务都处理完了,而其他队列里还有任务等待处理,干完活的线程如果干等着就显得很傻,老板也不愿意是不(ps:老板就喜欢压榨我们搬砖的),于是A就去其他线程的队列里窃取一个任务来执行。这里A和B(被窃取的线程)同时访问一个队列会出现竞争问题,那么为了避免这个问题,A(窃取线程)会从双端队列的尾部拿任务,B从头部拿任务。

    上面讲了理论,那么在执行的时候其实会有几个问题,比如

    1.双端队列只有一个任务时会出现竞争

    2.该算法会消耗更多的系统资源,比如创建多个线程和多个双端队列

    好处也显而易见:充分利用线程并行计算,减少线程间的竞争

    我们来看代码如果实现

    ForkJoinTask:ForkJoin任务类,它提供在任务中执行fork和join的操作机制,通常情况下我们不需要直接继承ForkJoinTask类,只需要继承它的子类就可以了,目前提供了两个子类

    RecursiveAction:用于没有返回结果的任务

    RecursiveTask:用于有返回结果的任务

    ForkJoinPool:ForkJoinTask需要通过ForkJoinTask来执行.

    下面看一个例子实现:

    首先创建一个任务类

    public class CountTask  extends RecursiveTask<Integer> {
        private static final int THRESHOLD=2;//阈值
        private  int start;
        private  int end;
        public  CountTask(int start,int end){
            this.start=start;
            this.end=end;
        }
    
        @Override
        protected Integer compute() {
            int sum=0;
            //如果任务足够小就计算任务
            boolean canCompute=(end-start)<=THRESHOLD;
            if(canCompute){
                for (int i=start;i<=end;i++){
                    sum+=i;
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    }catch (Exception e){
    
                    }
                }
            }else {
                int middle=(start+end)/2;
                CountTask leftTest=new CountTask(start,middle);
                CountTask rightTest=new CountTask(middle+1,end);
                //执行子任务
                leftTest.fork();
                rightTest.fork();
                //等待子任务执行完,并得到其结果
                int leftResult=leftTest.join();
                int rightResult=rightTest.join();
                //合并子任务
                sum=leftResult+rightResult;
            }
            return  sum;
        }
    }

      调用类:

    @RestController
    public class ForkJoinTest {
    
        private int start = 1;
        private int end = 4;
    
        @GetMapping("ForkJoinTest")
        public void forkjointest() throws InterruptedException, ExecutionException {
            ForkJoinPool forkJoinPool = new ForkJoinPool();
            //生成一个计算任务,负责计算1+2+3+4
            CountTask test = new CountTask(start, end);
            //执行任务
            Future<Integer> result = forkJoinPool.submit(test);
            System.out.println(result.get());
        }
    
        @GetMapping("test")
        public void test() throws InterruptedException {
            int sum = 0;
            for (int i = start; i <=end; i++) {
                sum += i;
                TimeUnit.SECONDS.sleep(1);
            }
            System.out.println(sum);
        }
    }

    这里写了两个方法,一个是通过ForkJoinTask来执行的,一个没有,我们执行看下耗时

    方法:forkjointest,参数:[Ljava.lang.Object;@6f277213
    10
    方法:forkjointest--cost:2009mills
    方法:test,参数:[Ljava.lang.Object;@a52eb2f
    10
    方法:test--cost:4001mills
  • 相关阅读:
    show variables 详解
    用expect做自动应答脚本
    percona-toolkit工具检查MySQL复制一致性及修复
    Discuz论坛架构改造
    注解处理器
    ora-12514 tns 监听程序当前无法识别连接描述符中请求的服务
    内嵌Tomcat和SpringMVC的简单整合测试
    @Import导入ImportSelector实现类demo
    Mybatis-Spring项目编写测试类
    @Configuration注解作用和在Spring中的处理
  • 原文地址:https://www.cnblogs.com/xwx20160804/p/11895698.html
Copyright © 2011-2022 走看看