【JUC】十、ForkJoin

妖狐艹你老母 2024-02-17 08:24 71阅读 0赞

#### 文章目录 ####

*  1、分支合并框架
 *  2、案例
 *  3、ForkJoinTask
 *  4、工作窃取算法
 *  5、ForkJoinPool

一个个任务执行在一个个线程上，倘若某一个任务耗时很久，期间其他线程都无事可做，显然没有利用好多核CPU这一计算机资源，因此，出现了"分而治之"的解决方案。

![在这里插入图片描述][d95393f46485479c8688368b6e0297d3.png]

## 1、分支合并框架 ##

Fork/Join即分支合并，用处是把一个大任务拆分成多个子任务来并行处理，最后再合并子任务的结果。（since Java1.7）

*  Fork：把一个复杂的任务进行拆分，大事化小
 *  Join：把拆分的子任务的结果做合并

示意图：

![在这里插入图片描述][051c6dc1d5444b6082ff2144e9c920a9.png]

举个实际例子：  
![在这里插入图片描述][3c59a1fa84a44d799883f44ec7ce5e06.png]

## 2、案例 ##

以下采用分支合并框架来实现1+2+…+100

![在这里插入图片描述][3ba1fb4ccdc24ab8a9562c2b30804d38.png]

通过继承RecursiveTask类自定义任务，泛型为result的类型：

//自定义任务类继承RecursiveTask
    class MyTask extends RecursiveTask<Integer> {
        
    
        //拆分差值不超过10，来计算10以内的运算
        private static final int VALUE = 10;
        //拆分起始值
        private int begin;
        //拆分结束值
        private int end;
        //返回结果
        private int result;
    
        //构造中传入起始值
        public MyTask(int begin, int end) {
        
            this.begin = begin;
            this.end = end;
        }
    
        //拆分与合并
        @Override
        protected Integer compute() {
        
            //判断即将要相加的两个数差值是否大于10
            if (end - begin <= VALUE) {
        
                for (int i = begin; i <= end; i++) {
        
                    result = result + i;
                }
            //否，则进一步拆分
            } else {
        
                //中间值
                int middle = (begin+end) /2;
                //左边
                MyTask task01 = new MyTask(begin, middle);
                //右边
                MyTask task02 = new MyTask(middle + 1, end);
                //调用方法拆分
                task01.fork();
                task02.fork();
                //合并结果
                result = task01.join() + task02.join();
    
            }
            return result;
        }
    }

以上自定义Task中体现的思路就是：问题足够小的时候，直接处理，否则就继续拆分。创建forkjoinpool对象，提交任务，获取执行结果：

public class ForkJoinDemo {
        
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
        
            //创建拆分任务对象
            MyTask myTask = new MyTask(0, 100);
            //创建分支合并池对象
            ForkJoinPool forkJoinPool = new ForkJoinPool();
            ForkJoinTask<Integer> forkJoinTask = forkJoinPool.submit(myTask);
            //获取最终合并之后的结果
            Integer result = forkJoinTask.get();
            System.out.println(result);
            //关闭池对象
            forkJoinPool.shutdown();
        }
    
    }

结果：

5050

## 3、ForkJoinTask ##

![在这里插入图片描述][ee7912e78234451ba21595a66e35ec26.png]  
ForkJoinTask有两个子类：Recursive即递归的意思

*  RecursiveTask：有返回值的任务
 *  RecursiveAction：没有返回值的任务

ForkJoinTask还实现了Future接口，即说明是一个可以异步获取结果的任务。常用方法有：

*  fork：分割子任务，判断当前线程是否可以转型为ForkJoinWorkerThread类型，即是不是通过ForkJoinPool线程池执行的ForkJoinTask，是就push到队列里

![在这里插入图片描述][b90dd0b73cf1430b9eddfdc42710a55c.png]

*  合并子任务结果：join

//调用方法拆分
    task01.fork();
    task02.fork();
    //合并结果
    result = task01.join() + task02.join();

*  获取任务执行结果：get，即等待任务执行完成并返回任务计算结果

[【方法源码详解参考这篇】][Link 1]

## 4、工作窃取算法 ##

先看类ForkJoinWorkerThread类，它是执行前面ForkJoinTask的线程。ForkJoin框架使用的线程维护了一个`双端队列`，并通过`工作窃取算法（work-stealing）`来提高任务执行的效率。

![在这里插入图片描述][2cb4a398f548424db284090b9a76bdef.png]

工作窃取算法，在ForkJoin框架中就是一个线程的任务队列里的活儿都干完了，就去其他线程维护的任务队列的尾部窃取一个任务来执行。用张网图示意：

![在这里插入图片描述][af0e2cfd55d34a6b87c23bd8ee535765.png]

因此：

*  对于线程自身维护的任务队列，线程自身执行任务的顺序是后进先出
 *  对于窃取其他线程队列的线程，则是先进先出（从队尾窃取并执行）

如图，对于线程1，任务的出入队都在队首，而负责窃取任务的线程2，其从队尾取任务，也就是说，只有线程1的任务队列只剩最后一个任务的时候，才会抢夺一次锁。如此，就大大提高了并发任务的执行效率。

## 5、ForkJoinPool ##

FoolJoinPool是运行ForkJoinTask的线程池，同ThreadPoolExecutor一样，也实现了Executor和ExecutorService接口

![在这里插入图片描述][ae54a8245b334666bdbd4351f4016e04.png]

任务提交到ForkJoinPool后，池里维护的线程来执行任务

ForkJoinPool forkJoinPool = new ForkJoinPool();
    ForkJoinTask<Integer> forkJoinTask = forkJoinPool.submit(myTask);

其构造方法有四个参数：

public ForkJoinPool(int parallelism,
                            ForkJoinWorkerThreadFactory factory,
                            UncaughtExceptionHandler handler,
                            boolean asyncMode){
        
    //.....

分别是：

*  parallelism：期望并发数，默认会使用Runtime.getRuntime().availableProcessors()的值，即当前计算机可用的CPU数量
 *  factory：创建ForkJoin工作线程的工厂，默认为defaultForkJoinWorkerThreadFactory
 *  handler：执行任务时遇到不可恢复的错误时的处理程序，默认为null
 *  asyncMode：工作线程获取任务使用FIFO（先进先出）模式还是LIFO模式，默认为LIFO（后进先出）

[d95393f46485479c8688368b6e0297d3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/3b74845581e14043a991887457b04ae1.png
[051c6dc1d5444b6082ff2144e9c920a9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/131c71455aaf455a9f195ac5a32d01c6.png
[3c59a1fa84a44d799883f44ec7ce5e06.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/fdb57e6114514b3eaaecf4c4f5d5ec86.png
[3ba1fb4ccdc24ab8a9562c2b30804d38.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/d3c32ad183bb429eaa4d7864096038ad.png
[ee7912e78234451ba21595a66e35ec26.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/536371bb628e4cb6b8b525fe9849cd2e.png
[b90dd0b73cf1430b9eddfdc42710a55c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/b2529cf80298479cb2c638b2ad021759.png
[Link 1]: https://blog.csdn.net/luanmousheng/article/details/78173342
[2cb4a398f548424db284090b9a76bdef.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/64bcad8f0f2c41bda7875815796467a4.png
[af0e2cfd55d34a6b87c23bd8ee535765.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/265dde6bed2349d1b6dbbd90472d12ec.png
[ae54a8245b334666bdbd4351f4016e04.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/c1946cba8ac440099e2876b093e6a21a.png