Fork/Join框架的使用

Fork/Join框架的使用

        Java7提供了ForkJoinPool来支持将一个任务拆分成多个“小任务”并行计算，再把多个“小任务”的结果合并成总的计算结果。ForkJoinPool类是Fork/Join框架的核心，它和ThreadPoolExecutor一样也是ExecutorService接口的实现类。

 

        ForkJoinPool最大的特殊之处就在于其实现了工作窃取（work-stealing）。所谓工作窃取，即当前线程的Task已经全被执行完毕，则自动从其他线程的任务池末尾取出任务继续执行。

 

        一个fork/join框架之下的任务由ForkJoinTask类表示。ForkJoinTask实现了Future接口，可以按照Future接口的方式来使用。在ForkJoinTask类中之重要的两个方法fork和join。fork方法用异步方式启动任务的执行，join方法则等待任务完成并返回结果。

 

        在创建任务时，通常不直接使用ForkJoinTask类，是使用它的两个抽象子类：

               RecursiveAction：没有返回值的任务

               RecursiveTask：有返回值的任务

 

对RecursiveAction抽象类的进一步封装：

public abstract class AbstractRecursiveAction<T> extends RecursiveAction {
	private T[] array; //数据数组
	private int start; //要处理的数据段的开始索引
	private int middle; //要处理的数据段的中间索引
	private int end; //要处理的数据段的结束索引
	private int threshold; //临界值：每个数据段的最大长度
    
    public AbstractRecursiveAction(T[] array, int start, int end, int threshold) {
        super();
        this.array = array;
        this.start = start;
        this.end = end;
        this.threshold = threshold;
        this.middle = start + (end - start) / 2;
    }
    
	@Override
	protected void compute() {
		if(notNeedRecursion()){
			handleFinalAction();
        }else{
        	handleRecursiveAction();
        }
	}
	
	public abstract void handleFinalAction();
	
	public abstract void handleRecursiveAction();
	
	/**
	 * 不需要递归拆分任务
	 */
	protected boolean notNeedRecursion(){
		return (end - start + 1) <= threshold;
	}

	protected Integer[] getLeftData() {
		Integer[] leftArray = new Integer[getMiddle() - getStart() + 1];
    	Arrays.stream(getArray(), getStart(), getMiddle()+1).collect(Collectors.toList()).toArray(leftArray);
        System.out.println(Arrays.toString(leftArray));
		return leftArray;
	}

	protected Integer[] getRightData() {
		Integer[] rightArray = new Integer[getEnd() - getMiddle()];
    	Arrays.stream(getArray(), getMiddle()+1, getEnd()+1).collect(Collectors.toList()).toArray(rightArray);
        System.out.println(Arrays.toString(rightArray));
		return rightArray;
	}

	public T[] getArray() {
		return array;
	}

	public int getStart() {
		return start;
	}

	public int getMiddle() {
		return middle;
	}

	public int getEnd() {
		return end;
	}

	public int getThreshold() {
		return threshold;
	}
}

 

对RecursiveTask抽象类的进一步封装：

public abstract class AbstractRecursiveTask<T, V> extends RecursiveTask<V> {
	private T[] array; //数据数组
	private int start; //要处理的数据段的开始索引
	private int middle; //要处理的数据段的中间索引
	private int end; //要处理的数据段的结束索引
	private int threshold; //临界值：每个数据段的最大长度
    
    public AbstractRecursiveTask(T[] array, int start, int end, int threshold) {
        super();
        this.array = array;
        this.start = start;
        this.end = end;
        this.threshold = threshold;
        this.middle = start + (end - start) / 2;
    }
    
	@Override
	protected V compute() {
		if(notNeedRecursion()){
			return handleFinalTask();
        }else{
        	return handleRecursiveTask();
        }
	}
	
	public abstract V handleFinalTask();
	
	public abstract V handleRecursiveTask();
	
	/**
	 * 不需要递归拆分任务
	 */
	protected boolean notNeedRecursion(){
		return (end - start + 1) <= threshold;
	}

	protected Integer[] getLeftData() {
		Integer[] leftArray = new Integer[getMiddle() - getStart() + 1];
    	Arrays.stream(getArray(), getStart(), getMiddle()+1).collect(Collectors.toList()).toArray(leftArray);
        System.out.println(Arrays.toString(leftArray));
		return leftArray;
	}

	protected Integer[] getRightData() {
		Integer[] rightArray = new Integer[getEnd() - getMiddle()];
    	Arrays.stream(getArray(), getMiddle()+1, getEnd()+1).collect(Collectors.toList()).toArray(rightArray);
        System.out.println(Arrays.toString(rightArray));
		return rightArray;
	}

	public T[] getArray() {
		return array;
	}

	public int getStart() {
		return start;
	}

	public int getMiddle() {
		return middle;
	}

	public int getEnd() {
		return end;
	}

	public int getThreshold() {
		return threshold;
	}
}

 

创建用于并行计算总和的RecursiveTask子类：

public class SumTask extends AbstractRecursiveTask<Integer, Integer> {
	public SumTask(Integer array[], int start, int end, int threshold){
        super(array, start, end, threshold);
	}
	
	@Override
	public Integer handleFinalTask() {
		int sum = 0;
		try{
			sum = Arrays.stream(getArray(), getStart(), getEnd()+1)
					.mapToInt(Integer::intValue)
					.sum();
			System.out.println(Thread.currentThread().getName() + " sum=" + sum);
		}catch(Exception ex){
			completeExceptionally(ex);
		}
		return sum;
	}
	
	@Override
	public Integer handleRecursiveTask() {
    	//left
    	Integer[] leftArray = getLeftData();
    	SumTask left = new SumTask(leftArray, 0, leftArray.length-1, getThreshold());
    	
    	//right
    	Integer[] rightArray = getRightData();
    	SumTask right = new SumTask(rightArray, 0, rightArray.length-1, getThreshold());
        
        //第一种用法
//    	right.fork(); //fork：将任务放到线程池中异步调度
//    	Integer leftValue = left.compute(); //compute：在当前线程执行任务
//    	Integer rightValue = right.join();  //join：等待并获取任务结果
//    	return leftValue + rightValue;
    	
    	//第二种用法
        //并行执行两个小任务
    	invokeAll(left, right);
    	return left.join() + right.join();
	}
}

 

测试代码：

private static Integer[] getData(int length){
	Integer[] array = new Integer[length];
	IntStream.rangeClosed(1, length)
		.boxed()
		.collect(Collectors.toList())
		.toArray(array);
	return array;
}

private static void handleSum(){
//	ForkJoinPool pool = ForkJoinPool.commonPool(); //最大工作线程数等于CPU核数-1
	ForkJoinPool pool = new ForkJoinPool(4); //指定处理任务的线程数
	Integer sum = 0;
	
	int length = 10;
	Integer[] array = getData(length);
	
	SumTask task = new SumTask(array, 0, array.length-1, 5);
	Future<Integer> future = pool.submit(task);
	
	sum = 0;
	try {
		sum = future.get();
	} catch (InterruptedException|ExecutionException e) {
		System.out.println(task.getException().toString());
	}

	System.out.println("sum=" + sum);
	
	pool.shutdown();
}