Fork/Join框架是Java 7提供的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。Fork/Join框架要完成两件事情：
1.任务分割：首先Fork/Join框架需要把大的任务分割成足够小的子任务，如果子任务比较大的话还要对子任务进行继续分割
2.执行任务并合并结果：分割的子任务分别放到双端队列里，然后启动几个线程分别从双端队列里获取任务执行。子任务执行完的结果都放在另外一个队列里，启动一个线程从队列里取数据，然后合并这些数据。
在Java的Fork/Join框架中，使用两个类完成上述操作
1.ForkJoinTask:我们要使用Fork/Join框架，首先需要创建一个ForkJoin任务。该类提供了在任务中执行fork和join的机制。通常情况下我们不需要直接集成ForkJoinTask类，只需要继承它的子类，Fork/Join框架提供了两个子类：
a.RecursiveAction：用于没有返回结果的任务
b.RecursiveTask:用于有返回结果的任务
2.ForkJoinPool:ForkJoinTask需要通过ForkJoinPool来执行
任务分割出的子任务会添加到当前工作线程所维护的双端队列中，进入队列的头部。当一个工作线程的队列里暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务(工作窃取算法)。
Fork/Join框架的实现原理
ForkJoinPool由ForkJoinTask数组和ForkJoinWorkerThread数组组成，ForkJoinTask数组负责将存放程序提交给ForkJoinPool，而ForkJoinWorkerThread负责执行这些任务。
example
public class ForkJoinRecursiveTask {
private final static int MAX_THRESHOLD = 200;
public static void main(String[] args) {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
ForkJoinTask<Integer> future = forkJoinPool.submit(new CalculateRecursiveTask(0, 1000));
try {
Integer result = future.get();
System.out.println(result);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
private static class CalculateRecursiveTask extends RecursiveTask<Integer>{
private final int start;
private final int end;
private CalculateRecursiveTask(int start, int end) {
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
if (end-start <= MAX_THRESHOLD){
return IntStream.rangeClosed(start,end).sum();
}else {
int middle = (start+end)/2;
CalculateRecursiveTask leftTask = new CalculateRecursiveTask(start, middle);
CalculateRecursiveTask rightTask = new CalculateRecursiveTask(middle+1,end);
leftTask.fork();
rightTask.fork();
return leftTask.join()+rightTask.join();
}
}
}
}
example
public class ForkJoinRecursiveActive {
private final static int MAX_THRESHOLD = 3;
private final static AtomicInteger SUM = new AtomicInteger(0);
public static void main(String[] args) throws InterruptedException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
forkJoinPool.submit(new CalculateRecursiveActive(0, 10));
forkJoinPool.awaitTermination(5, TimeUnit.SECONDS);
Optional.of(SUM).ifPresent(System.out::println);
}
private static class CalculateRecursiveActive extends RecursiveAction{
private final int start;
private final int end;
private CalculateRecursiveActive(int start, int end) {
this.start = start;
this.end = end;
}
@Override
protected void compute() {
if ((start+end)<MAX_THRESHOLD){
SUM.addAndGet(IntStream.rangeClosed(start,end).sum());
}else {
int middle = (start+end)/2;
CalculateRecursiveActive leftActive = new CalculateRecursiveActive(start,middle);
CalculateRecursiveActive rightActive = new CalculateRecursiveActive(middle+1,end);
leftActive.fork();
rightActive.fork();
}
}
}
}
