Fork\Join分而治之

分而治之"/>

Fork\Join分而治之

fork/join 分而治之

1.什么是Fork/join框架？

从JDK1.7开始，Java提供Fork/Join框架用于并行执行任务。它的思想就是讲一个大任务分割成若干小任务，最终汇总每个小任务的结果得到这个大任务的结果。如下图：

2.什么是分而治之思想

可以简单的理解为：将规模为N的问题，当N<阈值，直接解决；当N>阈值，将N分解为K个小规模子问题，子问题互相对立，与原问题形式相同，将子问题的解合并得到原问题的解。

3. 工作窃取（work-stealing）

• 假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。如果现场A的任务执行完成后，于是它就去其他线程B的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行,也就是线程B从头部拿取，而线程A从尾部获取任务开始执行。

• 工作窃取算法的优点是充分利用线程进行并行计算，并减少了线程间的竞争，其缺点是在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且消耗了更多的系统资源，比如创建多个线程和多个双端队列。

4.如何实现Fork/join

java 的ForkJoinTask 抽象类中提供compute方法给我们实现这种思想。java又提供两个抽象类继承ForkJoinTask ，分别是RecursiveAction和RecursiveTask.他们的区别在于：RecursiveTask有返回值，RecursiveAction无返回值。

4.1 fork/join标准规范

4.2 RecursiveTask的有返回值实现：

接下来我们以这样的例子，Fork/Join的同步用法实现有返回结果值：统计整形数组中所有元素的和

public class SumArray {private static class SumTask extends RecursiveTask<Integer>{private final static int THRESHOLD = MakeArray.ARRAY_LENGTH/10;private int[] src; //表示我们要实际统计的数组private int fromIndex;//开始统计的下标private int toIndex;//统计到哪里结束的下标//定义一个构造方法初始化参数public SumTask(int[] src, int fromIndex, int toIndex) {this.src = src;this.fromIndex = fromIndex;this.toIndex = toIndex;}@Overrideprotected Integer compute() {//小于阈值时，直接计算if(toIndex-fromIndex < THRESHOLD) {int count = 0;for(int i=fromIndex;i<=toIndex;i++) {count = count + src[i];}return count;}else {	//大于等于阈值时，拆分任务int mid = (fromIndex+toIndex)/2;SumTask left = new SumTask(src,fromIndex,mid);SumTask right = new SumTask(src,mid+1,toIndex);invokeAll(left,right);return left.join()+right.join();//任务结果返回}}}public static void main(String[] args) {//ForkJoinPool 是ExecutorService的实现类，负责工作线程的管理、任务队列的维护，以及控制整个任务调度流程；ForkJoinPool pool = new ForkJoinPool();int[] src = MakeArray.makeArray();//获取数组值方法，这里就不展示了SumTask innerFind = new SumTask(src,0,src.length-1);long start = System.currentTimeMillis();pool.invoke(innerFind);//同步调用System.out.println("Task is Running.....");//join方法是返回任务的结果System.out.println("The count is "+innerFind.join()+" spend time:"+(System.currentTimeMillis()-start)+"ms");}
}

4.2 RecursiveAction的无返回值实现：

例子是Fork/Join的异步用法不要求返回值，控制台打印结果：遍历指定目录（含子目录）寻找指定txt结尾的文件

public class FindDirsFiles extends RecursiveAction{private File path;//当前任务需要搜寻的目录public FindDirsFiles(File path) {this.path = path;}public static void main(String [] args){try {// 用一个 ForkJoinPool 实例调度总任务ForkJoinPool pool = new ForkJoinPool();FindDirsFiles task = new FindDirsFiles(new File("F:/"));pool.execute(task);//异步调用System.out.println("Task is Running......");Thread.sleep(1);//为了证明异步执行其他步骤int otherWork = 0;for(int i=0;i<100;i++){otherWork = otherWork+i;}System.out.println("Main Thread done sth......,otherWork="+otherWork);task.join();//阻塞的方法，等待异步完成System.out.println("Task end");} catch (Exception e) {e.printStackTrace();}}@Overrideprotected void compute() {List<FindDirsFiles> subTasks = new ArrayList<>();//定义子任务个数File[] files = path.listFiles();//获取目录所有文件和目录if(files!=null) {for(File file:files) {if(file.isDirectory()) {//如果是目录，就加入子任务中查询subTasks.add(new FindDirsFiles(file));}else {//遇到文件，检查if(file.getAbsolutePath().endsWith("txt")) {System.out.println("文件："+file.getAbsolutePath());}}}if(!subTasks.isEmpty()) {for(FindDirsFiles subTask:invokeAll(subTasks)) {subTask.join();//等待子任务执行完成}}}}
}

以上就是Fork/Join的简单实现分而治之思想。