其他分享
首页 > 其他分享> > Fork/Join 原理详解和使用

Fork/Join 原理详解和使用

作者:互联网

一、简介


从JDK1.7开始,Java提供ForkJoin框架用于并行执行任务,它的思想就是讲一个大任务分割成若干小任务,最终汇总每个小任务的结果得到这个大任务的结果。

 

1、ForkJoinPool

 

既然任务是被逐渐的细化的,那就需要把这些任务存在一个池子里面,这个池子就是ForkJoinPool,它与其它的ExecutorService区别主要在于它使用“工作窃取“,那什么是工作窃取呢?

 

一个大任务会被划分成无数个小任务,这些任务被分配到不同的队列,这些队列有些干活干的块,有些干得慢。于是干的快的,一看自己没任务需要执行了,就去隔壁的队列里面拿去任务执行。

 

2、ForkJoinTask

 

ForkJoinTask就是ForkJoinPool里面的每一个任务。他主要有两个子类:RecursiveAction和RecursiveTask。然后通过fork()方法去分配任务执行任务,通过join()方法汇总任务结果,

 

 

 

 

 

这就是整个过程的运用。他有两个子类,使用这两个子类都可以实现我们的任务分配和计算。

 

(1)RecursiveAction 一个递归无结果的ForkJoinTask(没有返回值)

 

(2)RecursiveTask 一个递归有结果的ForkJoinTask(有返回值)

 

ForkJoinPool由ForkJoinTask数组和ForkJoinWorkerThread数组组成,ForkJoinTask数组负责存放程序提交给ForkJoinPool的任务,而ForkJoinWorkerThread数组负责执行这些任务。

 

下面我们就来看看如何去使用。

 

二、使用

 

1、RecursiveTask :有返回结果

 

(1)第一步:创建MyRecursiveTask子类在ForkJoinTest中

private static class MyRecursiveTask extends RecursiveTask<Integer>{
        private final int start;
        private final int end;
        public MyRecursiveTask(int start, int end) {
            this.start = start;
            this.end = end;
        }
        @Override
        protected Integer compute() {
            //如果任务小的不能再细分了,那就直接计算
            if(end-start<=threshold) {
                //rangeClosed包含结束节点,是闭合的
                return IntStream.rangeClosed(start,end).sum();
            }else {
                int mid = (start + end ) /2;
                MyRecursiveTask leftTask = new MyRecursiveTask(start, mid);
                MyRecursiveTask rightTask = new MyRecursiveTask(mid+1, end);
                leftTask.fork();
                rightTask.fork();
                return leftTask.join() + rightTask.join() ;
            }   
        }
    }

 

在这个方法中我们传进去数据,然后使用二分法继续分配给子任务,当任务小的不能再分,那就汇总返回。

 

(2)第二步在ForkJoinTest中去测试

 

public class ForkJoinTest {
    private final static int threshold = 3;
    public static void main(String[] args) {
        final ForkJoinPool pool = new ForkJoinPool();
        ForkJoinTask<Integer> result = pool.submit(new MyRecursiveTask(0, 100));
        try {
            System.out.println(result.get());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

 





在这个类中我们定义了一个阈值,然后创建一个ForkJoinPool,在这个池子中新建我们刚刚创建的Task任务,最终返回我们结果

2、RecursiveAction:无返回结果

 

(1)第一步:创建MyRecursiveAction子类在ForkJoinTest中

private static class MyRecursiveAction extends RecursiveAction{
        private final int start;
        private final int end;  
        public MyRecursiveAction(int start, int end) {
            this.start = start;
            this.end = end;
        }
        @Override
        protected void compute() {
            //如果任务小的不能再细分了,那就直接计算
            if(end-start<=threshold) {
                //rangeClosed包含结束节点,是闭合的
                sum.addAndGet(IntStream.rangeClosed(start, end).sum());
            }else {
                int mid = (start + end ) /2;
                MyRecursiveTask leftTask = new MyRecursiveTask(start, mid);
                MyRecursiveTask rightTask = new MyRecursiveTask(mid+1, end);
                leftTask.fork();
                rightTask.fork();
            }
        }
    }

 

在这个方法中我们不需要有return语句。过程和之前的RecursiveTask类似。

 

(2)第二步在ForkJoinTest中去测试

 

public class ForkJoinTest {
    private final static int threshold = 3;
    private final static AtomicInteger sum = new AtomicInteger(0);
    public static void main(String[] args) throws InterruptedException {
        final ForkJoinPool pool = new ForkJoinPool();
        //注意在这里不需要有返回值了
        pool.submit(new MyRecursiveAction(0, 100));
        pool.awaitTermination(100, TimeUnit.MILLISECONDS);
        //我们依然还可以输出最终的返回值
        System.out.println(sum);
    }
}

 

ForkJoinTask在执行的时候可能会抛出异常,在主线程中无法直接获取,但可以通过ForkJoinTask提供的isCompletedAbnormally()方法来检查任务是否已经抛出异常或已经被取消。

标签:Fork,Join,int,ForkJoinPool,private,start,任务,详解,end
来源: https://www.cnblogs.com/buptlyh/p/16596907.html