Fork/Join框架详解 -- 基于JDK1.8

叁歲伎倆 2021-10-23 13:56 312阅读 0赞

**目录**

1、Fork/Join简介

（1）产生背景

（2）Fork/Join结构：

2、实例Demo：

3、原理及应用场景

--------------------

# 1、Fork/Join简介 #

## （1）产生背景 ##

摩尔定律（百度百科）：摩尔定律是由英特尔（Intel）创始人之一戈登·摩尔（Gordon Moore）提出来的。其内容为：当价格不变时，集成电路上可容纳的元器件的数目，约每隔18-24个月便会增加一倍，性能也将提升一倍。换言之，每一美元所能买到的电脑性能，将每隔18-24个月翻一倍以上。这一定律揭示了信息技术进步的速度。

尽管这种趋势已经持续了超过半个世纪，摩尔定律仍应该被认为是观测或推测，而不是一个物理或自然法。预计定律将持续到至少2015年或2020年。

由于技术的限制，单个CPU上面的元器件数目趋于稳定，摩尔定律的失效，所以在单个CPU上面的性能提升得到限制；

那么要提升软件的运行效率，就开始从多核CPU上面找方法，一台服务器搭载多个CPU运行；

## （2）Fork/Join结构： ##

IDEA查看结构图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70][]

从并发到并行：涉及到并发与并行的概念，可以参考前一篇转载帖子：

[https://blog.csdn.net/lejustdoit/article/details/99210251][https_blog.csdn.net_lejustdoit_article_details_99210251]

基本思想：把一个大的问题划分成若干个小的子任务，分而治之，最后合并子任务以解决原问题；

步骤1：分割原问题，首先需要把大任务分割成子任务，有可能子任务还是比较大，所以还需要不停地分割，直到分割出的子任务足够小。

步骤2：求解子任务，分割的子任务分别放在双端队列里，然后几个启动线程分

别从双端队列里获取任务执行。子任务执行完的结果都统一放在一个队列里，启动一个线程从队列里拿数据，然后合并这些数据。

在分治法当中，子问题一般是相互独立的，因此可以通过递归的思想来求解子问题；我们较为熟悉分治思想常见有：二分查找法、快速排序、汉诺塔问题等；

Fork/Join框架是Java 7开始提供的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。Fork（分开）、Join（合并）；

下图来自于《Java并发编程的艺术》—方腾飞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 1][]

# 2、实例Demo： #

ForkJoinTask是Fork/Join框架的实现者，通过fork()方法：将任务拆分入队并异步执行；join()方法：等待计算结果（子类是否支持来返回结果）；

但他本身又是一个抽象类，从上图的结构我们可以看到，ForkJoinTask还有3个子类实现类，这3个子类也是抽象类，我们在实现的时候，可以根据自己的业务场景，各自继承这3个子类：

RecursiveAction-无返回值的任务，

RecursiveTask-有返回值的任务，

CountedCompleter-完成任务后将触发其它任务；

并且在实际应用中，需要通过结合Fork/Join框架的线程池ForkJoinPool执行使用，下面我们给出一个例子，求和1~100累加和，当每个子任务求累积和的数值大于10个的时候，就进行拆分，直到小于10个，这里我们的计算结果有返回值，所以采取了继承RecursiveTask类，重写compute方法：

import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import java.util.concurrent.RecursiveTask;
    
    /**
     * 多任务分解Demo
     * Fork/Join框架测试
     * 20190813
     * Ethan
     */
    public class TestForkJoin {
    
        private  static final Integer num = 10;
    
        static class MyRecursiveTask extends RecursiveTask<Integer> {
    
            private Integer lowValue;
            private Integer highValue;
    
            @Override
            protected Integer compute() {
                //不满足执行子任务的条件，继续拆分成两个子任务
                if (highValue-lowValue> num){
                    //拆分的第一个子任务，继续递归判断
                    MyRecursiveTask task1 = new MyRecursiveTask(lowValue,(lowValue+highValue)/2);
                    task1.fork();
                    //拆分的第二个子任务，继续递归判断
                    MyRecursiveTask task2 = new MyRecursiveTask((lowValue+highValue)/2+1,highValue);
                    task2.fork();
                    return task1.join()+task2.join();
                }else//满足条件，开始执行
                {
                    System.out.println("start compute: lowValue:"+lowValue+";highValue:"+highValue);
                    Integer sum = 0;
                    for (int i = lowValue; i <=highValue; i++) {
                        sum+=i;
                    }
                    return sum;
                }
            }
            public MyRecursiveTask(Integer lowValue,Integer highValue){
                this.lowValue = lowValue;
                this.highValue = highValue;
            }
        }
    
        public static void main(String[] args) {
            ForkJoinPool pool = new ForkJoinPool();
            //线程池提交任务
            ForkJoinTask task= pool.submit(new MyRecursiveTask(1,100));
            try {
                //输出各个子任务求和的总值
                System.out.println("totalSum:"+task.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }
    
    =======================================
    计算结果：
    start compute: lowValue:1;highValue:7
    start compute: lowValue:8;highValue:13
    start compute: lowValue:14;highValue:19
    start compute: lowValue:76;highValue:82
    start compute: lowValue:83;highValue:88
    start compute: lowValue:89;highValue:94
    start compute: lowValue:95;highValue:100
    start compute: lowValue:64;highValue:69
    start compute: lowValue:70;highValue:75
    start compute: lowValue:26;highValue:32
    start compute: lowValue:33;highValue:38
    start compute: lowValue:39;highValue:44
    start compute: lowValue:45;highValue:50
    start compute: lowValue:58;highValue:63
    start compute: lowValue:20;highValue:25
    start compute: lowValue:51;highValue:57
    totalSum:5050

# 3、原理及应用场景 #

上面的排序例子可能不太能说明Fork/Join框架的实际应用场景性能。

作为对 Fork/Join 型线程池的实现，是ExecutorService的实现类，因此是一种特殊的线程池。创建实例后，调用submit()或者invoke()来执行指定任务。

注：

ForkJoinPool 相比于 ThreadPoolExecutor，还有一个非常重要的特点（优点）在于，ForkJoinPool具有 Work-Stealing （工作窃取）的能力。

所谓 Work-Stealing，在 ForkJoinPool中的实现为：线程池中每个线程都有一个互不影响的任务队列（双端队列），线程每次都从自己的任务队列的队头中取出一个任务来运行；如果某个线程对应的队列已空并且处于空闲状态，而其他线程的队列中还有任务需要处理但是该线程处于工作状态，那么空闲的线程可以从其他线程的队列的队尾取一个任务来帮忙运行 —— 感觉就像是空闲的线程去偷人家的任务来运行一样，所以叫 “工作窃取”。

Work-Stealing 的适用场景是不同的任务的耗时相差比较大，即某些任务需要运行较长时间，而某些任务会很快的运行完成，这种情况下用 Work-Stealing 很合适；但是如果任务的耗时很平均，则此时 Work-Stealing 并不适合，因为窃取任务时不同线程需要抢占锁，这可能会造成额外的时间消耗，而且每个线程维护双端队列也会造成更大的内存消耗。所以 ForkJoinPool 并不是 ThreadPoolExecutor 的替代品，而是作为对 ThreadPoolExecutor 的补充。

ForkJoinPool 和 ThreadPoolExecutor 都是 ExecutorService（线程池），但ForkJoinPool 的独特点在于：

ThreadPoolExecutor 只能执行 Runnable 和 Callable 任务，而 ForkJoinPool 不仅可以执行 Runnable 和 Callable 任务，还可以执行 Fork/Join 型任务 —— ForkJoinTask —— 从而满足并行地实现分治算法的需要；

ThreadPoolExecutor 中任务的执行顺序是按照其在共享队列中的顺序来执行的，所以后面的任务需要等待前面任务执行完毕后才能执行，而 ForkJoinPool 每个线程有自己的任务队列，并在此基础上实现了 Work-Stealing 的功能，使得在某些情况下 ForkJoinPool 能更大程度的提高并发效率。

ForkJoinPool构造函数

parallelism：可并行级别，Fork/Join框架将依据这个并行级别的设定，决定框架内并行执行的线程数量。并行的每一个任务都会有一个线程进行处理，但是千万不要将这个属性理解成Fork/Join框架中最多存在的线程数量，也不要将这个属性和ThreadPoolExecutor线程池中的corePoolSize、maximumPoolSize属性进行比较，因为ForkJoinPool的组织结构和工作方式与后者完全不一样。而后续的讨论中，读者还可以发现Fork/Join框架中可存在的线程数量和这个参数值的关系并不是绝对的关联（有依据但并不全由它决定）。

factory：当Fork/Join框架创建一个新的线程时，同样会用到线程创建工厂。只不过这个线程工厂不再需要实现ThreadFactory接口，而是需要实现ForkJoinWorkerThreadFactory接口。后者是一个函数式接口，只需要实现一个名叫newThread的方法。在Fork/Join框架中有一个默认的ForkJoinWorkerThreadFactory接口实现：DefaultForkJoinWorkerThreadFactory。

handler：异常捕获处理器。当执行的任务中出现异常，并从任务中被抛出时，就会被handler捕获。

asyncMode：这个参数也非常重要，从字面意思来看是指的异步模式，它并不是说Fork/Join框架是采用同步模式还是采用异步模式工作。Fork/Join框架中为每一个独立工作的线程准备了对应的待执行任务队列，这个任务队列是使用数组进行组合的双向队列。即是说存在于队列中的待执行任务，即可以使用先进先出的工作模式，也可以使用后进先出的工作模式。

如果你对Fork/Join框架没有特定的执行要求，可以直接使用不带有任何参数的构造函数。也就是说推荐基于当前操作系统可以使用的CPU内核数作为Fork/Join框架内最大并行任务数量，这样可以保证CPU在处理并行任务时，尽量少发生任务线程间的运行状态切换（实际上单个CPU内核上的线程间状态切换基本上无法避免，因为操作系统同时运行多个线程和多个进程）。

以上大部分内容参考自：

[https://blog.csdn.net/yinwenjie/article/details/71524140][https_blog.csdn.net_yinwenjie_article_details_71524140]

如果对源码感兴趣，可以参考：

[https://www.jianshu.com/p/32a15ef2f1bf][https_www.jianshu.com_p_32a15ef2f1bf]

[https://www.jianshu.com/p/6a14d0b54b8d][https_www.jianshu.com_p_6a14d0b54b8d]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70]: /images/20211023/e7a21f274f3149bf8e88ecef727caa32.png
[https_blog.csdn.net_lejustdoit_article_details_99210251]: https://blog.csdn.net/lejustdoit/article/details/99210251
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 1]: /images/20211023/36c55b209c7e4bb4b09d120a5f18749d.png
[https_blog.csdn.net_yinwenjie_article_details_71524140]: https://blog.csdn.net/yinwenjie/article/details/71524140
[https_www.jianshu.com_p_32a15ef2f1bf]: https://www.jianshu.com/p/32a15ef2f1bf
[https_www.jianshu.com_p_6a14d0b54b8d]: https://www.jianshu.com/p/6a14d0b54b8d