工作窃取模式 Fork/Join

小咪咪 2022-12-29 09:54 186阅读 0赞

### **介绍：** ###

Fork：分割任务

Join：合并结果

当一个大任务（比较耗时，数据量比较大）时，可以把这个任务分割成若干互不依赖的子任务，为了减少线程竞争，把子任务分别放入不同队列，并为每个队列创建一个单独的工作线程，当有线程提前把自己队列任务做完，为了提高效率，已经做完任务的线程会去其他队列窃取任务执行，以帮助其他未完成的线程（而不是自己等着）。此时他们访问同一个队列，为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列。被窃取任务线程永远从双端队列头部取任务执行，窃取任务线程永远从双端队列尾部拿任务执行

### **优缺点：** ###

优点：就是充分利用线程进行并行计算，减少了线程间的竞争

缺点：当双端队列里只有一个任务时，还时会存在竞争。该算法创建多个线程和多个双端队列会消耗更过的系统资源

### **实现：** ###

**ForkJoinTask：**要使用ForkJoin框架，首先必须创建一个ForkJoin任务，它提供在任务中执行fork()和join()的操作机制，一般情况下继承ForkJoinTask的子类并重写其compute方法即可

1）RecursiveAction：用于没有返回结果的任务

2）RecursiveTask：用于有返回结果的任务

**ForkJoinPool：**ForkJoinTask需要通过ForkJoinPool类执行

public class ForkJoinTest extends RecursiveTask<Integer> {
        //临界值
        private static final long THURSHOLD = 1000;
        private int start;
        private int end;
    
        public ForkJoinTest(int start, int end) {
            this.start = start;
            this.end = end;
        }
    
        @Override
        protected Integer compute() {
            int length = end - start;
            //如果任务分割的足够小就计算任务
            if (length <= THURSHOLD) {
                int sum = 0;
                for (int i = start; i <= end; i++) {
                    sum += i;
                }
                return sum;
            } else {
                //任务大于阈值，分割成两个子任务
                int middle = (start + end) / 2;
                //创建子任务
                ForkJoinTest left = new ForkJoinTest(start, middle);
                ForkJoinTest right = new ForkJoinTest(middle + 1, end);
                //执行子任务
                left.fork();
                right.fork();
                //等待子任务执行完并结果合并
                Integer leftResult = left.join();
                Integer rightResult = right.join();
                return leftResult + rightResult;
            }
        }
    
        public static void main(String[] args) throws Exception {
            ForkJoinPool pool = new ForkJoinPool();
            //计算  0,10000的总和
            ForkJoinTask<Integer> task = new ForkJoinTest(0, 10000);
            ForkJoinTask<Integer> result = pool.submit(task);
            System.out.println(result.get());
        }
    }

### **异常处理：** ###

ForkJoinTask在执行时可能抛出异常，但无法再主线程里捕捉异常，可以通过ForkJoinTask的isCompletedAbnnormally()方法来检查任务是否已经抛出异常或已经被取消，调用getException()可以获取异常，代码如下：

if(task.isCompletedAbnormally()){
        System.out.println(task.getException());
    }

### **ForkJoinPool**与ThreadPool的区别： ###

ThreadPool：只有“外部任务”，也就是调用者放到队列里的任务

ForkJoinPool：有“外部任务”，还有“内部任务”，也就是任务自身在执行过程中，分裂出”子任务“，递归，再次放入队列