Java多线程之Fork/Join

Java中并行计算的框架就是Fork/Join，这个框架的原理是将一个很大的任务分解成无数个小任务，完成每个小任务后，将其计算结果再合并处理，得到任务的最终答案。

首先，这个框架有一个核心的线程池ForkJoinPool，这个线程池实现了工作窃取算法，这个名气很专业，其实就是当一个工作线程完成了属于自己的任务后，再去主动寻找其他线程还没完成的任务，这样就可以提高CPU利用率。这个线程池一般在任务中只需要一个，因此我们使用时，通常定义为static的。

public final static ForkJoinPool mainPool = new ForkJoinPool();

其次，任务定义统一使用了ForkJoinTask类，这个类有两个子类，一个是RecursiveAction,一个是RecursiveTask，前者没有返回值，后者的计算可以携带返回值。我们一般在使用时，集成这两个子类，实现其compute方法即可。

先看一个不带返回值的例子，有一个20个元素的数组，我们需要给这个数组中每个元素做一个加1的操作。直接干代码，一般按照这个模式来写就好了。

package com.lenovo.plm.dms.p26;

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

public class ForkJoinTest {
public final static ForkJoinPool mainPool = new ForkJoinPool();

public static void main(String[] args) {
int n=20;
int[] a = new int
;
for(int i=0;i<n;i++){
a[i] = i;
}
SubTask subTask = new SubTask(a,0,n);
mainPool.invoke(subTask);
for(int i=0;i<n;i++){
System.out.println(a[i]+" ");;
}
}
}

class SubTask extends RecursiveAction{

/** */
private static final long serialVersionUID = 1L;

private int[] a;
private int begin;
private int end;

public SubTask(int[] a,int begin,int end){
this.a = a;
this.begin = begin;
this.end = end;
}

@Override
protected void compute() {
// TODO Auto-generated method stub
if(end - begin >10){
int mid = (end+begin)/2;
SubTask sub1 = new SubTask(a,begin,mid);
SubTask sub2 = new SubTask(a,mid,end);
invokeAll(sub1,sub2);
}else{
for(int i=begin;i<end;i++){
a[i] = a[i]+1;
}
}
}
}

再看一个带返回值的例子，有一个100个元素的数组，我们要计算这些元素的总和。直接干代码：

package com.lenovo.plm.dms.p26;

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

public class ForkJoinTest2 {
public final static ForkJoinPool mainPool = new ForkJoinPool();

public static void main(String[] args) {
int n = 100;
int[] a = new int
;
long sum = 0;
for(int i=0;i<100;i++){
a[i] = i+1;
sum = sum+a[i];
}
SubTask1 task = new SubTask1(a,0,100);
int count = mainPool.invoke(task);
System.out.println(count+"-----------"+sum);
}
}

class SubTask1 extends RecursiveTask<Integer>{

/** */
private static final long serialVersionUID = 1L;

private int[] a;
private int begin;
private int end;

public SubTask1(int[] a,int begin,int end){
this.a = a;
this.begin = begin;
this.end = end;
}

@Override
protected Integer compute() {
// TODO Auto-generated method stub
int result = 0;
if(end-begin>10){
int mid = (begin+end)/2;
SubTask1 t1 = new SubTask1(a,begin,mid);
SubTask1 t2 = new SubTask1(a,mid,end);
invokeAll(t1,t2);
try {
result = t1.get() + t2.get();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (ExecutionException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}else{
for(int i=begin;i<end;i++){
result = result+a[i];
}
}
return result;
}

}

粗暴之后，说一句重点，就是在实现compute方法时，需要自己写逻辑来完成任务分割，因此使用这个框架的重点是如何分割任务。后续我还会来分析这个框架的源代码，看看工作窃取是怎么窃取的。

另外，这个框架的使用很少有书籍，之前有听说过这个框架的名字，因此我在网上搜索了一下，写这个文章是我的学习笔记，参考了如下文章：

http://blog.csdn.net/andycpp/article/details/8903155

http://www.importnew.com/14506.html