Fork/Join使用学习

JDK7提供了一个将任务“分而治之”的框架 — Fork/Join。它把一个大的任务分割成足够小的子任务，如果子任务比较大的话还要对子任务进行继续分割。分割的子任务分别放到双端队列里，然后启动线程分别从双端队列里获取任务执行。子任务执行完的结果都放在另外一个队列里，启动一个线程从队列里取数据，然后合并这些数据。

Fork/Join的思想如下所示：

RecursiveTask

RecursiveTask适用于将任务分而治之，并且有返回值的情况，举个计算1到100和的例子：

publicclassRecursiveTest{//定义最小区间为10privatefinalstaticintMAX_THRESHOLD=10;publicstaticvoidmain(String[]args){finalForkJoinPoolforkJoinPool=newForkJoinPool();ForkJoinTask<Integer>future=forkJoinPool.submit(newCalculateRecursiveTask(1,100));try{Integerresult=future.get();System.out.println(result);}catch(InterruptedException|ExecutionExceptione){e.printStackTrace();}}privatestaticclassCalculateRecursiveTaskextendsRecursiveTask<Integer>{//起始privateintstart;//结束privateintend;publicCalculateRecursiveTask(intstart,intend){this.start=start;this.end=end;}@OverrideprotectedIntegercompute(){//如果起始和结束范围小于我们定义的区间范围，则直接计算if((end-start)<=MAX_THRESHOLD){returnIntStream.rangeClosed(start,end).sum();}else{//否则，将范围一分为二，分成两个子任务intmiddle=(start+end)/2;CalculateRecursiveTaskleftTask=newCalculateRecursiveTask(start,middle);CalculateRecursiveTaskrightTask=newCalculateRecursiveTask(middle+1,end);//执行子任务leftTask.fork();rightTask.fork();//汇总子任务returnleftTask.join()+rightTask.join();}}}}

ForkJoinPool使用submit或invoke提交的区别：invoke是同步执行，调用之后需要等待任务完成，才能执行后面的代码；submit是异步执行，只有在Future调用get的时候会阻塞。

启动程序输出如下：

5050

其实这里执行子任务调用fork方法并不是最佳的选择，最佳的选择是invokeAll方法：

//执行子任务//leftTask.fork();//rightTask.fork();invokeAll(leftTask,rightTask);//汇总子任务returnleftTask.join()+rightTask.join();

RecursiveAction

使用方式和RecursiveTask类似，只不过没有返回值：

publicclassRecursiveActionTest{//定义最小区间为10privatefinalstaticintMAX_THRESHOLD=10;privatefinalstaticAtomicIntegerSUM=newAtomicInteger(0);publicstaticvoidmain(String[]args)throwsInterruptedException{finalForkJoinPoolforkJoinPool=newForkJoinPool();forkJoinPool.submit(newCalculateRecursiveAction(0,100));forkJoinPool.awaitTermination(2,TimeUnit.SECONDS);System.out.println(SUM);}privatestaticclassCalculateRecursiveActionextendsRecursiveAction{//起始privatefinalintstart;//结束privatefinalintend;privateCalculateRecursiveAction(intstart,intend){this.start=start;this.end=end;}@Overrideprotectedvoidcompute(){//如果起始和结束范围小于我们定义的区间范围，则直接计算if((end-start)<=MAX_THRESHOLD){SUM.addAndGet(IntStream.rangeClosed(start,end).sum());}else{//否则，将范围一分为二，分成两个子任务intmiddle=(end+start)/2;CalculateRecursiveActionleftAction=newCalculateRecursiveAction(start,middle);CalculateRecursiveActionrightAction=newCalculateRecursiveAction(middle+1,end);//执行子任务invokeAll(leftAction,rightAction);//没有汇总子任务结果过程，因为没有返回值。}}}}

输出结果也是5050。

理解Fork/Join框架API

Fork/Join框架在java.util.concurrent包下被实现。它的核心有4个类：

ForkJoinTask: 这是一个抽象任务类，并且运行在ForkJoinPool中。

ForkJoinPool:这是一个线程池管理并运行众多ForkJoinTask任务。

RecursiveAction: ForkJoinTask的子类，这个类没有返回值。

RecursiveTask: ForkJoinTask的子类，有返回值。

基本上，我们解决问题的代码是在RecursiveAction或者RecursiveTask中进行的，然后将任务提交由ForkJoinPool`执行，ForkJoinPool处理从线程管理到多核处理器的利用等各种事务。

我们先来理解一下这些类中的关键方法。

ForkJoinTask

这是一个运行在ForkJoinPool中的抽象的任务类。类型V指定了任务的返回结果。ForkJoinTask是一个类似线程的实体，它表示任务的轻量级抽象，而不是实际的执行线程。该机制允许由ForkJoinPool中的少量实际线程管理大量任务。其关键方法是：

final ForkJoinTaskfork()

final V join()

final V invoke()

fork()方法提交并执行异步任务，该方法返回ForkJoinTask并且调用线程继续运行。

join()方法等待任务直到返回结果。

invoke()方法是组合了fork()和join()，它开始一个任务并等待结束返回结果。

此外，ForkJoinTask中还提供了用于一次调用多个任务的两个静态方法

static void invokeAll(ForkJoinTask<?> task1, ForkJoinTask<?> task2) :执行两个任务

static void invokeAll(ForkJoinTask<?>… taskList):执行任务集合

RecursiveAction

这是一个递归的ForkJoinTask子类，不返回结果。Recursive意思是任务可以通过分治策略分成自己的子任务（在下面的下一节中，您将看到如何划分代码示例）。

我们必须重写compute()方法,并将计算代码写在其中：

protectedabstractvoidcompute();

RecursiveTask

和RecursiveAction一样，但是RecursiveTask有返回结果，结果类型由V指定。我们仍然需要重写compute()方法：

protectedabstractVcompute();

ForkJoinPool

这是Fork/Join框架的核心类。它负责线程的管理和ForkJoinTask的执行，为了执行ForkJoinTask，首先需要获取到ForkJoinPool的实例。

有两种构造器方式可以获取ForkJoinPool的实例，第一种使用构造器创建：

ForkJoinPool(): 使用默认的构造器创建实例，该构造器创建出的池与系统中可用的处理器数量相等。

ForkJoinPool(int parallelism):该构造器指定处理器数量，创建具有自定义并行度级别的池，该级别的并行度必须大于0，且不超过可用处理器的实际数量。

并行性的级别决定了可以并发执行的线程的数量。换句话说，它决定了可以同时执行的任务的数量——但不能超过处理器的数量。

但是，这并不限制池可以管理的任务的数量。ForkJoinPool可以管理比其并行级别多得多的任务。

获取ForkJoinPool实例的第二种方法是使用以下ForkJoinPool的静态方法获取公共池实例:

publicstaticForkJoinPoolcommonPool();

这种方式创建的池不受shutdown()或者shutdownNow()方法的影响，但是他会在System.exit()时会自动中止。任何依赖异步任务处理的程序在主体程序中止前都应该调用awaitQuiescence()方法。该方式是静态的，可以自动被使用。

作窃取算法

工作窃取（work-stealing）算法是指某个线程从其他队列里窃取任务来执行。

那么为什么需要使用工作窃取算法呢？假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。

工作窃取算法的优点是充分利用线程进行并行计算，并减少了线程间的竞争，其缺点是在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且消耗了更多的系统资源，比如创建多个线程和多个双端队列。

什么时候用

上面只是为了演示Fork/Join的用法，实际是采用这种方式计算反而更加费时，因为切割任务，分配线程需要额外的开销。其实什么时候用不必太纠结，一个足够大的任务，如果采用Fork/Join来处理比传统处理方式快的话，那就毫不犹豫的选择它吧！