多线程 ForkJoinPool

背景：ForkJoinPool的优势在于，可以充分利用多cpu，多核cpu的优势，把一个任务拆分成多个“小任务”，把多个“小任务”放到多个处理器核心上并行执行；当多个“小任务”执行完成之后，再将这些执行结果合并起来即可。这种思想值得学习。

Java7 提供了ForkJoinPool来支持将一个任务拆分成多个“小任务”并行计算，再把多个“小任务”的结果合并成总的计算结果。

ForkJoinPool是ExecutorService的实现类，因此是一种特殊的线程池。

使用方法：创建了ForkJoinPool实例之后，就可以调用ForkJoinPool的submit(ForkJoinTask<T> task) 或invoke(ForkJoinTask<T> task)方法来执行指定任务了。

其中ForkJoinTask代表一个可以并行、合并的任务。ForkJoinTask是一个抽象类，它还有两个抽象子类：RecusiveAction和RecusiveTask。其中RecusiveTask代表有返回值的任务，而RecusiveAction代表没有返回值的任务。

下面的UML类图显示了ForkJoinPool、ForkJoinTask之间的关系：

举例

以还行没有返回值的“大任务”（简单低打印1~300的数值）为例，程序将一个“大任务”拆分成多个“小任务”，并将任务交给ForkJoinPool来执行

package forkjoinpool;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.TimeUnit;

public class PrintTask extends RecursiveAction {
    private static final int THRESHOLD=50; //最多只能打印50个数
    private int start;
    private int end;

    public PrintTask(int start, int end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected void compute() {
      if(end-start<THRESHOLD){
       for(int i=start;i<=end;i++){
           System.out.println(Thread.currentThread().getName()+"的i的值"+i);
       }
      }else{
          int middle=(start+end)/2;
          PrintTask leftTask=new PrintTask(this.start,middle);
          PrintTask rightTask=new PrintTask(middle,this.end);
          leftTask.fork();
          rightTask.fork();

      }
    }

    public static void main(String[] args) throws InterruptedException {
        PrintTask printTask=new PrintTask(0,300);
        ForkJoinPool forkJoinPool=new ForkJoinPool();
        forkJoinPool.submit(printTask);
        forkJoinPool.awaitTermination(2, TimeUnit.MINUTES);
        forkJoinPool.shutdown();
    }
}

执行结果：

通过RecursiveTask的返回值，来对一个长度为100的数组元素进行累加。

package forkjoinpool;

import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;

public class Sum2Task extends RecursiveTask<Integer> {
    private static final  int THRESHOLD=20;
    private int array[];
    private int start;
    private int end;

    public Sum2Task(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Integer compute() {
        int sum=0;
     if(end-start<THRESHOLD){
         for(int i=start;i<end;i++){
          sum+=array[i];
         }
         return sum;
     }else{
         int middle=(start+end)/2;
         Sum2Task left=new Sum2Task(array,start,middle);
         Sum2Task right=new Sum2Task(array,middle,end);
         left.fork();
         right.fork();
         return left.join()+right.join();
     }

    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        int[] array=new int[100];
        Random random = new Random();
        int total =0;
        //初始化100个数组元素
        for(int i=0;i<100;i++){
            int temp = random.nextInt(20);
            //对数组元素赋值，并将数组元素的值添加到sum总和中
            total += (array[i]=temp);
        }
        System.out.println("初始化数组总和："+total);
        Sum2Task task = new Sum2Task(array,0,100);
        ForkJoinPool pool=ForkJoinPool.commonPool();
        Future<Integer> future =pool.submit(task); //提交分解的SumTask 任务
        System.out.println("多线程执行结果："+future.get());
        pool.shutdown(); //关闭线程池
    }
}

在Java 7中引入了一种新的线程池：ForkJoinPool。

它同ThreadPoolExecutor一样，也实现了Executor和ExecutorService接口。它使用了一个无限队列来保存需要执行的任务，而线程的数量则是通过构造函数传入，如果没有向构造函数中传入希望的线程数量，那么当前计算机可用的CPU数量会被设置为线程数量作为默认值。

ForkJoinPool主要用来使用分治法(Divide-and-Conquer Algorithm)来解决问题。典型的应用比如快速排序算法。

这里的要点在于，ForkJoinPool需要使用相对少的线程来处理大量的任务。

比如要对1000万个数据进行排序，那么会将这个任务分割成两个500万的排序任务和一个针对这两组500万数据的合并任务。以此类推，对于500万的数据也会做出同样的分割处理，到最后会设置一个阈值来规定当数据规模到多少时，停止这样的分割处理。比如，当元素的数量小于10时，会停止分割，转而使用插入排序对它们进行排序。

那么到最后，所有的任务加起来会有大概2000000+个。问题的关键在于，对于一个任务而言，只有当它所有的子任务完成之后，它才能够被执行。

所以当使用ThreadPoolExecutor时，使用分治法会存在问题，因为ThreadPoolExecutor中的线程无法像任务队列中再添加一个任务并且在等待该任务完成之后再继续执行。而使用ForkJoinPool时，就能够让其中的线程创建新的任务，并挂起当前的任务，此时线程就能够从队列中选择子任务执行。

以上程序的关键是fork()和join()方法。在ForkJoinPool使用的线程中，会使用一个内部队列来对需要执行的任务以及子任务进行操作来保证它们的执行顺序。

那么使用ThreadPoolExecutor或者ForkJoinPool，会有什么性能的差异呢？

首先，使用ForkJoinPool能够使用数量有限的线程来完成非常多的具有父子关系的任务，比如使用4个线程来完成超过200万个任务。但是，使用ThreadPoolExecutor时，是不可能完成的，因为ThreadPoolExecutor中的Thread无法选择优先执行子任务，需要完成200万个具有父子关系的任务时，也需要200万个线程，显然这是不可行的。

ps：ForkJoinPool在执行过程中，会创建大量的子任务，导致GC进行垃圾回收，这些是需要注意的。