写在后面

在JDK中，提供了这样一种性能：它可能将简单的逻辑拆分成一个个简略的逻辑来并行执行，待每个并行执行的逻辑执行实现后，再将各个后果进行汇总，得出最终的后果数据。有点像Hadoop中的MapReduce。

ForkJoin是由JDK1.7之后提供的多线程并发解决框架。ForkJoin框架的根本思维是分而治之。什么是分而治之？分而治之就是将一个简单的计算，依照设定的阈值分解成多个计算，而后将各个计算结果进行汇总。相应的，ForkJoin将简单的计算当做一个工作，而合成的多个计算则是当做一个个子工作来并行执行。

注：文章已同步收录到：https://github.com/sunshinely... 和 https://gitee.com/binghe001/t... 。如果文件对你有点帮忙，别忘记给个Star哦！如果小伙伴们有任何疑难，都能够加我微信【sun_shine_lyz】进行交换哦！

Java并发编程的倒退

对于Java语言来说，生来就反对多线程并发编程，在并发编程畛域也是在一直倒退的。Java在其倒退过程中对并发编程的反对越来越欠缺也正好印证了这一点。

  • Java 1 反对thread，synchronized。
  • Java 5 引入了 thread pools， blocking queues, concurrent collections，locks, condition queues。
  • Java 7 退出了fork-join库。
  • Java 8 退出了 parallel streams。

并发与并行

并发和并行在实质上还是有所区别的。

并发

并发指的是在同一时刻，只有一个线程可能获取到CPU执行工作，而多个线程被疾速的轮换执行，这就使得在宏观上具备多个线程同时执行的成果，并发不是真正的同时执行，并发能够应用下图示意。

并行

并行指的是无论何时，多个线程都是在多个CPU外围上同时执行的，是真正的同时执行。

分治法

根本思维

把一个规模大的问题划分为规模较小的子问题，而后分而治之，最初合并子问题的解失去原问题的解。

步骤

①宰割原问题；

②求解子问题；

③合并子问题的解为原问题的解。

咱们能够应用如下伪代码来示意这个步骤。

if(工作很小）{    间接计算失去后果}else{    分拆成N个子工作    调用子工作的fork()进行计算    调用子工作的join()合并计算结果}

在分治法中，子问题个别是互相独立的，因而，常常通过递归调用算法来求解子问题。

典型利用

  • 二分搜寻
  • 大整数乘法
  • Strassen矩阵乘法
  • 棋盘笼罩
  • 合并排序
  • 疾速排序
  • 线性工夫抉择
  • 汉诺塔

ForkJoin并行处理框架

ForkJoin框架概述

Java 1.7 引入了一种新的并发框架—— Fork/Join Framework，次要用于实现“分而治之”的算法，特地是分治之后递归调用的函数。

ForkJoin框架的实质是一个用于并行执行工作的框架， 可能把一个大工作宰割成若干个小工作，最终汇总每个小工作后果后失去大工作的计算结果。在Java中，ForkJoin框架与ThreadPool共存，并不是要替换ThreadPool

其实，在Java 8中引入的并行流计算，外部就是采纳的ForkJoinPool来实现的。例如，上面应用并行流实现打印数组元组的程序。

public class SumArray {    public static void main(String[] args){        List<Integer> numberList = Arrays.asList(1,2,3,4,5,6,7,8,9);        numberList.parallelStream().forEach(System.out::println);    }}

这段代码的背地就应用到了ForkJoinPool。

说到这里，可能有读者会问：能够应用线程池的ThreadPoolExecutor来实现啊？为什么要应用ForkJoinPool啊？ForkJoinPool是个什么鬼啊？！ 接下来，咱们就来答复这个问题。

ForkJoin框架原理

ForkJoin框架是从jdk1.7中引入的新个性,它同ThreadPoolExecutor一样，也实现了Executor和ExecutorService接口。它应用了一个有限队列来保留须要执行的工作，而线程的数量则是通过构造函数传入，如果没有向构造函数中传入指定的线程数量，那么以后计算机可用的CPU数量会被设置为线程数量作为默认值。

ForkJoinPool次要应用分治法(Divide-and-Conquer Algorithm)来解决问题。典型的利用比方疾速排序算法。这里的要点在于，ForkJoinPool可能应用绝对较少的线程来解决大量的工作。比方要对1000万个数据进行排序，那么会将这个工作宰割成两个500万的排序工作和一个针对这两组500万数据的合并工作。以此类推，对于500万的数据也会做出同样的宰割解决，到最初会设置一个阈值来规定当数据规模到多少时，进行这样的宰割解决。比方，当元素的数量小于10时，会进行宰割，转而应用插入排序对它们进行排序。那么到最初，所有的工作加起来会有大略200万+个。问题的关键在于，对于一个工作而言，只有当它所有的子工作实现之后，它才可能被执行。

所以当应用ThreadPoolExecutor时，应用分治法会存在问题，因为ThreadPoolExecutor中的线程无奈向工作队列中再增加一个工作并在期待该工作实现之后再继续执行。而应用ForkJoinPool就可能解决这个问题，它就可能让其中的线程创立新的工作，并挂起以后的工作，此时线程就可能从队列中抉择子工作执行。

那么应用ThreadPoolExecutor或者ForkJoinPool，性能上会有什么差别呢？

首先，应用ForkJoinPool可能应用数量无限的线程来实现十分多的具备父子关系的工作，比方应用4个线程来实现超过200万个工作。然而，应用ThreadPoolExecutor时，是不可能实现的，因为ThreadPoolExecutor中的Thread无奈抉择优先执行子工作，须要实现200万个具备父子关系的工作时，也须要200万个线程，很显然这是不可行的，也是很不合理的！！

工作窃取算法

如果咱们须要做一个比拟大的工作，咱们能够把这个工作宰割为若干互不依赖的子工作，为了缩小线程间的竞争，于是把这些子工作别离放到不同的队列里，并为每个队列创立一个独自的线程来执行队列里的工作，线程和队列一一对应，比方A线程负责解决A队列里的工作。然而有的线程会先把本人队列里的工作干完，而其余线程对应的队列里还有工作期待解决。干完活的线程与其等着，不如去帮其余线程干活，于是它就去其余线程的队列里窃取一个工作来执行。而在这时它们会拜访同一个队列，所以为了缩小窃取工作线程和被窃取工作线程之间的竞争，通常会应用双端队列，被窃取工作线程永远从双端队列的头部拿工作执行，而窃取工作的线程永远从双端队列的尾部拿工作执行。

工作窃取算法的长处：
充分利用线程进行并行计算，并缩小了线程间的竞争。

工作窃取算法的毛病：
在某些状况下还是存在竞争，比方双端队列里只有一个工作时。并且该算法会耗费更多的系统资源，比方创立多个线程和多个双端队列。

Fork/Join框架局限性：

对于Fork/Join框架而言，当一个工作正在期待它应用Join操作创立的子工作完结时，执行这个工作的工作线程查找其余未被执行的工作，并开始执行这些未被执行的工作，通过这种形式，线程充分利用它们的运行工夫来进步应用程序的性能。为了实现这个指标，Fork/Join框架执行的工作有一些局限性。

（1）工作只能应用Fork和Join操作来进行同步机制，如果应用了其余同步机制，则在同步操作时，工作线程就不能执行其余工作了。比方，在Fork/Join框架中，使工作进行了睡眠，那么，在睡眠期间内，正在执行这个工作的工作线程将不会执行其余工作了。
（2）在Fork/Join框架中，所拆分的工作不应该去执行IO操作，比方：读写数据文件。
（3）工作不能抛出查看异样，必须通过必要的代码来进去这些异样。

ForkJoin框架的实现

ForkJoin框架中一些重要的类如下所示。

ForkJoinPool 框架中波及的次要类如下所示。

1.ForkJoinPool类

实现了ForkJoin框架中的线程池，由类图能够看出，ForkJoinPool类实现了线程池的Executor接口。

咱们也能够从下图中看出ForkJoinPool的类图关系。

其中，能够应用Executors.newWorkStealPool()办法创立ForkJoinPool。

ForkJoinPool中提供了如下提交工作的办法。

public void execute(ForkJoinTask<?> task)public void execute(Runnable task)public <T> T invoke(ForkJoinTask<T> task)public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task)public <T> ForkJoinTask<T> submit(Callable<T> task)public <T> ForkJoinTask<T> submit(Runnable task, T result)public ForkJoinTask<?> submit(Runnable task)

2.ForkJoinWorkerThread类

实现ForkJoin框架中的线程。

3.ForkJoinTask<V>类

ForkJoinTask封装了数据及其相应的计算，并且反对细粒度的数据并行。ForkJoinTask比线程要轻量，ForkJoinPool中大量工作线程可能运行大量的ForkJoinTask。

ForkJoinTask类中次要包含两个办法fork()和join()，别离实现工作的分拆与合并。

fork()办法相似于Thread.start()，然而它并不立刻执行工作，而是将工作放入工作队列中。跟Thread.join()办法不同，ForkJoinTask的join()办法并不简略的阻塞线程，而是利用工作线程运行其余工作，当一个工作线程中调用join()，它将解决其余工作，直到留神到指标子工作曾经实现。

咱们能够应用下图来示意这个过程。

ForkJoinTask有3个子类：

  • RecursiveAction：无返回值的工作。
  • RecursiveTask：有返回值的工作。
  • CountedCompleter：实现工作后将触发其余工作。

4.RecursiveTask<V> 类

有返回后果的ForkJoinTask实现Callable。

5.RecursiveAction类

无返回后果的ForkJoinTask实现Runnable。

6.CountedCompleter<T> 类

在工作实现执行后会触发执行一个自定义的钩子函数。

ForkJoin示例程序

package io.binghe.concurrency.example.aqs; import lombok.extern.slf4j.Slf4j;import java.util.concurrent.ForkJoinPool;import java.util.concurrent.Future;import java.util.concurrent.RecursiveTask;@Slf4jpublic class ForkJoinTaskExample extends RecursiveTask<Integer> {    public static final int threshold = 2;    private int start;    private int end;    public ForkJoinTaskExample(int start, int end) {        this.start = start;        this.end = end;    }    @Override    protected Integer compute() {        int sum = 0;        //如果工作足够小就计算工作        boolean canCompute = (end - start) <= threshold;        if (canCompute) {            for (int i = start; i <= end; i++) {                sum += i;            }        } else {            // 如果工作大于阈值，就决裂成两个子工作计算            int middle = (start + end) / 2;            ForkJoinTaskExample leftTask = new ForkJoinTaskExample(start, middle);            ForkJoinTaskExample rightTask = new ForkJoinTaskExample(middle + 1, end);             // 执行子工作            leftTask.fork();            rightTask.fork();             // 期待工作执行完结合并其后果            int leftResult = leftTask.join();            int rightResult = rightTask.join();             // 合并子工作            sum = leftResult + rightResult;        }        return sum;    }    public static void main(String[] args) {        ForkJoinPool forkjoinPool = new ForkJoinPool();         //生成一个计算工作，计算1+2+3+4        ForkJoinTaskExample task = new ForkJoinTaskExample(1, 100);         //执行一个工作        Future<Integer> result = forkjoinPool.submit(task);         try {            log.info("result:{}", result.get());        } catch (Exception e) {            log.error("exception", e);        }    }}

重磅福利

微信搜一搜【冰河技术】微信公众号，关注这个有深度的程序员，每天浏览超硬核技术干货，公众号内回复【PDF】有我筹备的一线大厂面试材料和我原创的超硬核PDF技术文档，以及我为大家精心筹备的多套简历模板（不断更新中），心愿大家都能找到心仪的工作，学习是一条时而郁郁寡欢，时而开怀大笑的路，加油。如果你通过致力胜利进入到了心仪的公司，肯定不要懈怠放松，职场成长和新技术学习一样，逆水行舟。如果有幸咱们江湖再见！

另外，我开源的各个PDF，后续我都会继续更新和保护，感激大家长期以来对冰河的反对！！