浅谈Fork/Join的使用

最近正在学习Java多线程编程，偶然间看见Fork/Join框架，特此记录下来以备查看。

Fork/Join框架，简单来说，就是将一个大的任务拆分成若干的小的任务，最后再讲各个小任务的执行结果进行Join汇总，得到大任务的结果。

举个例子，现在有一个累加的需求，要求从1加到100000，按照传统的单线程方式从1+2+3+…+100000，可以想象，这样的执行效率并不高，这个时候我们就可以尝试拆分成若干个小任务，1+…+10000,10001+…+20000，依次类推，这个时候我们有10个小任务并发执行，再将每一个小任务执行的结果相加，就得到了1加到100000的值，可以想象，这样一来效率会提高很多，这就是Fork/Join框架的基本思想。

对于Fork/Join来说，提供了两个类RecursiveTask和RecursiveAction。我们在使用时，只需要根据具体业务继承这两个类的其一，然后重写compute（）方法即可，区别是RecursiveTask的compute方法有返回值而RecursiveAction的compute方法有返回值没有返回值。下面以两个小例子展示一下两种方式的具体用法

RecursiveTask

现需要统计一个目录下面的文件个数。我们可以进行一个简单的分析，目录下面可能有子目录和文件，子目录下面可能还会有子目录，这个时候可能各位看官都有自己一个想法来实现这个功能，那我们用RecursiveTask怎么实现呢，话不多说，直接上代码。

    public class SumFilesTask extends RecursiveTask<Integer>{
    	
    	private File path;
    	
    	public SumFilesTask(File path) {
    		this.path = path;
    	}
    
    	@Override
    	protected Integer compute() {
    		int count = 0; //计数器，统计文件个数
    		
    		File[] files = path.listFiles();//得到一个路径下的所有文件和文件夹
    		
    		List<SumFilesTask> sumFilesTasks = new ArrayList<>();//子任务的集合
    		
    		if(files!=null) {
    			for(File file:files) { //遍历文件
    				if(file.isDirectory()) {//如果是目录
    					sumFilesTasks.add(new SumFilesTask(file));//开一个子任务，将任务加到子任务集合中
    				}else {
    					count++;//当前目录文件数加一
    				}
    			}
    		}
    		if(!sumFilesTasks.isEmpty()) { //如果子任务集合不为空
    			for(SumFilesTask sumFilesTask:invokeAll(sumFilesTasks)) {
    				/*
    				 * sumFilesTask.join()，每一个子任务执行，join方法是阻塞方法，
    				 * 要当前子任务执行完成才会执行下一个任务
    				 */
    				count= count + sumFilesTask.join(); //将子任务统计结果汇总
    			}
    		}
    		return count;
    	}
    	
    	public static void main(String[] args) {
    		
    		SumFilesTask sumFilesTask = new SumFilesTask(new File("D:/threadtool/"));//D盘threadtool文件夹
    		ForkJoinPool forkJoinPool = new ForkJoinPool();
    		forkJoinPool.invoke(sumFilesTask);//同步方法
    		sumFilesTask.join();//保证主任务完成
    		System.out.println("目录D:/threadtool/以及其子目录下文件总数为:"+count);
    	}
    }

运行结果如下：

RecursiveAction

查找一个目录下所有.txt文件，代码如下：

public class FindFilesTask extends RecursiveAction{
	
	private File path;
	
	public FindFilesTask(File path) {
		this.path = path;
	}
	@Override
	protected void compute() {
		
		File[] files = path.listFiles();//得到一个路径下的所有文件和文件夹
		
		List<FindFilesTask> findFilesTasks = new ArrayList<>();//子任务的集合
		
		if(files!=null) {
			for(File file:files) { //遍历文件
				if(file.isDirectory()) {//如果是目录
					findFilesTasks.add(new FindFilesTask(file));//开一个子任务，将任务加到子任务集合中
				}else {
					if(file.getAbsolutePath().endsWith("txt")){//如果是txt文件，打印路径
						System.out.println(file.getAbsolutePath());
					}
				}
			}
		}
		if(!findFilesTasks.isEmpty()) { //如果子任务集合不为空
			for(FindFilesTask findFilesTask:invokeAll(findFilesTasks)) {
				/*
				 * sumFilesTask.join()，每一个子任务执行，join方法是阻塞方法，
				 * 要当前子任务执行完成才会执行下一个任务
				 */
				findFilesTask.join(); //将子任务统计结果汇总
			}
		}
	}
	
	public static void main(String[] args) throws InterruptedException {
		
		FindFilesTask findFilesTask = new FindFilesTask(new File("D:/threadtool/"));//D盘threadtool文件夹
		ForkJoinPool forkJoinPool = new ForkJoinPool();
		//forkJoinPool.invoke(findFilesTask);此方法为同步方法，同步方法也可以实现，此处我们另外验证一下异步方法execute
		forkJoinPool.execute(findFilesTask);//异步方法
		/**
		 * 为了验证异步，我们先让主线程sleep 1ms
		 */
		Thread.sleep(1);
		System.out.println("I am ok");
		findFilesTask.join();//保证主任务完成
		System.out.println("Task end");
	}
}