concurrent之ForkJoin初识笔记

程序员文章站 2022-05-04 21:39:36

...

需要先认识 RecursiveTask

RecursiveTask：递归任务

描述：有一个fork类来把大任务分割成子任务，有可能子任务还是很大，所以还需要不停的分割，直到分割出的子任务足够小。

Fork/Join框架要完成两件事情：
　　1.任务分割：首先Fork/Join框架需要把大的任务分割成足够小的子任务，如果子任务比较大的话还要对子任务进行继续分割
　　2.执行任务并合并结果：分割的子任务分别放到双端队列里，然后几个启动线程分别从双端队列里获取任务执
行。子任务执行完的结果都放在另外一个队列里，启动一个线程从队列里取数据，然后合并这些数据。
　　在Java的Fork/Join框架中，使用两个类完成上述操作
　　1.ForkJoinTask:我们要使用Fork/Join框架，首先需要创建一个ForkJoin任务。该类提供了在任务中执行
fork和join的机制。通常情况下我们不需要直接集成ForkJoinTask类，只需要继承它的子类，Fork/Join框架
提供了两个子类：
　　　　a.RecursiveAction：用于没有返回结果的任务
　　　　b.RecursiveTask:用于有返回结果的任务
　　2.ForkJoinPool:ForkJoinTask需要通过ForkJoinPool来执行
　　任务分割出的子任务会添加到当前工作线程所维护的双端队列中，进入队列的头部。当一个工作线程的队列里
暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务(工作窃取算法)。

代码栗子：

package com.geek.forkjoin;

import com.geek.threadandjuc.base.CountTask;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;

public class ForkJoin1 extends  RecursiveTask<Integer> {

        private static final int THREAD_HOLD = 2;

        private int start;
        private int end;

        public ForkJoin1(int start, int end) {
            this.start = start;
            this.end = end;
        }

        @Override
        protected Integer compute() {
            int sum = 0;
            //如果任务足够小就计算
            boolean canCompute = (end - start) <= THREAD_HOLD;
            if (canCompute) {
                for (int i = start; i <= end; i++) {
                    sum += i;
                }
            } else {
                int middle = (start + end) / 2;
                ForkJoin1 left = new ForkJoin1(start, middle);
                ForkJoin1 right = new ForkJoin1(middle + 1, end);
                //执行子任务
                left.fork();
                right.fork();
                //获取子任务结果
                int lResult = left.join();
                int rResult = right.join();
                sum = lResult + rResult;
            }
            return sum;
        }

        public static void main(String[] args) {
            ForkJoinPool pool = new ForkJoinPool();
            ForkJoin1 task = new ForkJoin1(1, 4);
            Future<Integer> result = pool.submit(task);
            try {
                System.out.println(result.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
    }
}

concurrent之ForkJoin初识笔记