一、ForkJoin
ForkJoin是由JDK1.7后提供多线并发处理框架。ForkJoin的框架的基本思想是分而治之。什么是分而治之？分而治之就是将一个复杂的计算，按照设定的阈值进行分解成多个计算，然后将各个计算结果进行汇总。相应的ForkJoin将复杂的计算当做一个任务。而分解的多个计算则是当做一个子任务。
二、ForkJoin的使用
- 一、 创建Task
使用ForkJoin框架，需要创建一个ForkJoin的任务，而ForkJoinTask是一个抽象类，我们不需要去继承ForkJoinTask进行使用。因为ForkJoin框架为我们提供了RecursiveAction和RecursiveTask。我们只需要继承ForkJoin为我们提供的抽象类的其中一个并且实现compute方法。
private static class SumTask extends RecursiveTask<Integer> {
private int threshold ;
private static final int segmentation = 10;
private int[] src;
private int fromIndex;
private int toIndex;
public SumTask(int formIndex,int toIndex,int[] src){
this.fromIndex = formIndex;
this.toIndex = toIndex;
this.src = src;
this.threshold = src.length/segmentation;
}
@Override
protected Integer compute() {
if((toIndex - fromIndex)<threshold ){
int count = 0;
System.out.println(" from index = "+fromIndex
+" toIndex="+toIndex);
for(int i = fromIndex;i<=toIndex;i++){
count+=src[i];
}
return count;
}else{
int mid = (fromIndex+toIndex)/2;
SumTask left = new SumTask(fromIndex,mid,src);
SumTask right = new SumTask(mid+1,toIndex,src);
invokeAll(left,right);
return left.join()+right.join();
}
}
}
- 二、使用ForkJoinPool进行执行
task要通过ForkJoinPool来执行，分割的子任务也会添加到当前工作线程的双端队列中，
进入队列的头部。当一个工作线程中没有任务时，会从其他工作线程的队列尾部获取一个任务(工作窃取)。
public static void main(String[] args) { int[] array = MakeArray.createIntArray(); ForkJoinPool forkJoinPool= new ForkJoinPool(); SumTask sumTask = new SumTask(0,array.length-1,array); long start = System.currentTimeMillis(); forkJoinPool.invoke(sumTask); System.out.println("The count is "+sumTask.join() +" spend time:"+(System.currentTimeMillis()-start)+"ms"); }
三、RecursiveTask和RecursiveAction区别
- RecursiveTask
通过源码的查看我们可以发现RecursiveTask在进行exec之后会使用一个result的变量进行接受返回的结果。而result返回结果类型是通过泛型进行传入。也就是说RecursiveTask执行后是有返回结果。
public abstract class RecursiveTask<V> extends ForkJoinTask<V> { private static final long serialVersionUID = 5232453952276485270L; /** * The result of the computation. */ V result; /** * The main computation performed by this task. * @return the result of the computation */ protected abstract V compute(); public final V getRawResult() { return result; } protected final void setRawResult(V value) { result = value; } /** * Implements execution conventions for RecursiveTask. */ protected final boolean exec() { result = compute(); return true; } }
- RecursiveAction
RecursiveAction在exec后是不会保存返回结果，因此RecursiveAction与RecursiveTask区别在与RecursiveTask是有返回结果而RecursiveAction是没有返回结果。
public abstract class RecursiveAction extends ForkJoinTask<Void> { private static final long serialVersionUID = 5232453952276485070L; /** * The main computation performed by this task. */ protected abstract void compute(); /** * Always returns {@code null}. * * @return {@code null} always */ public final Void getRawResult() { return null; } /** * Requires null completion value. */ protected final void setRawResult(Void mustBeNull) { } /** * Implements execution conventions for RecursiveActions. */ protected final boolean exec() { compute(); return true; } }
ForkJoinTask是RecursiveAction与RecursiveTask的父类， ForkJoinTask中使用了模板模式进行设计
,将ForkJoinTask的执行相关的代码进行隐藏，通过提供抽象类暴露用户的实际业务处理。
三、ForJoin注意点
使用ForkJoin将相同的计算任务通过多线程的进行执行。从而能提高数据的计算速度。在google的中的大数据处理框架mapreduce就通过类似ForkJoin的思想。通过多线程提高大数据的处理。但是我们需要注意：
- 使用这种多线程带来的数据共享问题，在处理结果的合并的时候如果涉及到数据共享的问题，我们尽可能使用JDK为我们提供的并发容器。
- 在使用JVM的时候我们要考虑OOM的问题，如果我们的任务处理时间非常耗时，并且处理的数据非常大的时候。会造成OOM。
- ForkJoin也是通过多线程的方式进行处理任务。那么我们不得不考虑是否应该使用ForkJoin。因为当数据量不是特别大的时候，我们没有必要使用ForkJoin。因为多线程会涉及到上下文的切换。所以数据量不大的时候使用串行比使用多线程快。
四、ForkJoin工作窃取（work-stealing）
为什么ForkJoin会存在工作窃取呢？因为我们将任务进行分解成多个子任务的时候。每个子任务的处理时间都不一样。例如分别有子任务AB。如果子任务A的1ms的时候已经执行，子任务B还在执行。那么如果我们子任务A的线程等待子任务B完毕后在进行汇总，那么子任务A线程就会在浪费执行时间，最终的执行时间就以最耗时的子任务为准。而如果我们的子任务A执行完毕后，处理子任务B的任务，并且执行完毕后将任务归还给子任务B。这样就可以提高执行效率。而这种就是工作窃取。
五、ForkJoin排序
public class SortForkJoin { /** * 数组排序 * * @param arry * @return */ public static int[] sort(int[] arry) { if (arry.length == 0) return arry; for (int index = 0; index < arry.length - 1; index++) { int pre_index = index; int currentValue = arry[index + 1]; while (pre_index >= 0 && arry[pre_index] > currentValue) { arry[pre_index + 1] = arry[pre_index]; pre_index--; } arry[pre_index + 1] = currentValue; } return arry; } /** * 组合 * * @param left * @param right * @return */ public static int[] merge(int[] left, int[] right) { int[] result = new int[left.length + right.length]; for (int resultIndex = 0, leftIndex = 0, rightIndex = 0; resultIndex < result.length; resultIndex++) { if (leftIndex >= left.length) { result[resultIndex] = right[rightIndex++]; } else if (rightIndex >= right.length) { result[resultIndex] = left[leftIndex++]; } else if (left[leftIndex] > right[rightIndex]) { result[resultIndex] = right[rightIndex++]; } else { result[resultIndex] = left[leftIndex++]; } } return result; } static class SortTask extends RecursiveTask<int[]> { private int threshold; private int start; private int end; private int segmentation ; private int[] src; public SortTask(int[] src,int start,int end,int segmentation){ this.src = src; this.start = start; this.end = end; this.threshold = src.length/segmentation; this.segmentation = segmentation; } @Override protected int[] compute() { if((end - start) <threshold){ int mid = (end-start)/2; SortTask leftTask = new SortTask(src,start,mid,segmentation); SortTask rightTask = new SortTask(src,mid+1,end,segmentation); invokeAll(leftTask,rightTask); return SortForkJoin.merge(leftTask.join(),rightTask.join()); }else{ return SortForkJoin.sort(src); } } } @Test public void test() { int[] array = MakeArray.createIntArray(); ForkJoinPool forkJoinPool= new ForkJoinPool(); SortTask sortTask =new SortTask(array,0,array.length-1,1000); long start = System.currentTimeMillis(); forkJoinPool.execute(sortTask); System.out.println( " spend time:"+(System.currentTimeMillis()-start)+"ms"); } }