Fork/Join
Fork/Join框架
Fork/Join框架是Java 7提供的并行执行任务框架。思想是将大任务分成小任务,然后小任务分别计算出结果,再结合起来,汇总结果作为大任务结果。其思想和MapReduce的思想非常类似。对于任务的分割,要求各个子任务之间相互独立,能够并行独立地执行任务,互相之间不影响。
从以上描述可以看出,Fork/Join框架主要由子任务、任务调度两部分组成。
-
ForkJoinPool: ForkJoin同样是利用了线程池,和ThreadPoolExecutor一样实现了自己的线程池。对任务进行调度
-
ForkJoinTask: 在Fork/Join框架中执行的任务类。子任务对其进行继承
- RecursiveAction: 用于不用返回结果的子任务
- RecursiveTask: 用于返回有结果的子任务
ForkJoinPool
ForkJoinPool是ForkJoin框架中的任务调度器,提供了三种调度子任务的方法
- execute 异步执行指定任务
- invoke/invokeAll 执行指定任务,等待完成,返回结果
- submit 异步执行指定任务,并立即返回一个Future对象
ForkJoin提供工作窃取(work-stealing)算法,使得若一个工作线程的task队列为空,没有任务可以执行时,从其他工作线程中获取任务主动执行。为了实现工作窃取,在工作线程中维护双端队列,窃取任务线程从队尾取任务,被窃取任务线程从队头获取任务。这种机制充分利用线程进行并行计算,减少线程竞争。但是当队列中只存在一个任务了时,两个线程去取反而会造成资源浪费。
Fork/Join维护两个数组,分别是ForkJoinTask数组和ForkJoinWorkerThread数组。ForkJoinTask负责存放程序交给ForkJoinPool的任务,ForkJoinWorkerThread负责执行这些任务。
ForkJoinTask
通过定义ForkJoinTask来执行任务,一般不需要直接继承ForkJoinTask,可以根据是否需要返回计算结果,选择继承RecursiveAction还是RecursiveTask。继承后需要重写compute方法,以完成任务分割以及计算结果汇总的定义。
-
fork()
当ForkJoinTask提交执行任务时,会调用fork()方法。fork会让ForkJoinWorkerThread调用push来异步执行该任务。
-
join()
join() 方法是阻塞当前进程,等待取得结果.在join的具体实现中,会利用Future的特性查看当前任务的状态来判断返回结果。
一般开发ForkJoin程序,主要是实现自定义ForkJoinTask。
Demo
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
public class testForkJoin{
public static void main(String[] args)
{
ForkJoinPool pool = new ForkJoinPool();
MyForkJoinTask task = new MyForkJoinTask(1,100);
Future<Integer> result = pool.submit(task);
System.out.println("result is" + result.get());
}
}
Task
import java.util.concurrent.RecursiveTask;
public class MyForkJoinTask extends RecursiveTask<Integer>{
private final int spilSize = 2;
private int start,end;
public MyForkJoinTask (int start,int end)
{
this.start=start;
this.end=end;
}
protected Integer compute() {
// TODO Auto-generated method stub
int sum = 0;
if((end-start)<2)
{
for(int i=start;i<end;i++)
{
sum+=i;
}
}
else
{
int middle = (start+end)/2;
MyForkJoinTask firstTask = new MyForkJoinTask(start,middle);
MyForkJoinTask secondTask = new MyForkJoinTask(middle+1,end);
firstTask.fork(); //提交任务
secondTask.fork(); //
Integer firstResult = firstTask.join(); //阻塞线程等待任务结果
Integer secondResult = secondTask.join()
sum=firstResult.intValue()+secondResult.intValue();
}
return null;
}
}