Fork/Join 框架简介及简易使用

#什么是 Fork/Join 框架

Fork/ Join 框架是 Java 7 开始提供的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。

#什么是工作窃取算法

在了解 Fork/Join 框架之前必须还要了解一下工作窃取算法，因为 Fork/Join 框架主要是在使用这个算法来进行工作的。
工作窃取算法（work-stealing）主要是指在同一个并行计算的诸多线程中，一个线程在完成自己的任务之后，窃取（协助）其他未完成任务的线程，进而提升整个任务的的完成效率。在工作窃取算法中，每个线程都会使用双端队列， 被窃取任务线程永远 从双端队列的头部拿任务执行， 而窃取任务的线程永远从双端队列的尾部拿任务执行。

#Fork/Join 框架的执行原理

• 步骤 1： 分割任务：。 首先需要有一个 fork 类来把大任务 分割成子任务， 有可能子任务还是很大， 所以还需要不停地分割，直到 分割出的子任务足够小。
• 步骤 2：执行任务并合并结果： 分割的子任务分别放在双端队列 里， 然后几个启动线程分别从双端队列里获取任务执行。子任务执行完的结果都统一放在一个队列里，启动一个线程从队列里拿数据，然后合并这些数据。

#如何使用 Fork/Join 框架

public class CountTask extends RecursiveTask<Integer> {
	private static final int THRESHOLD = 2;
	// 阈值
	private int start;
	private int end;

	public CountTask(int start, int end) {
		this.start = start;
		this.end = end;
	}

	@Override
	protected Integer compute() {
		int sum = 0;
		// 如果任务足够小就 计算任务
		boolean canCompute = (end - start) <= THRESHOLD;
		if (canCompute) {
			for (int i = start; i <= end; i++) {
				sum += i;
			}
		} else {
			// 如果任务大于 阈值，就分裂成 两个子任务计算
			int middle = (start + end) / 2;
			CountTask leftTask = new CountTask(start, middle);
			CountTask rightTask = new CountTask(middle + 1, end);
			// 执行子任务
			leftTask.fork();
			rightTask.fork();
			// 等待子任务 执行完， 并得到 其 结果
			int leftResult = leftTask.join();
			int rightResult = rightTask.join();
			// 合并子任务
			sum = leftResult + rightResult;
		}
		return sum;
	}

	public static void main(String[] args) {
		ForkJoinPool forkJoinPool = new ForkJoinPool();
		// 生成一个计算任务， 负责计算1+...=100
		CountTask task = new CountTask(1, 100);
		// 执行一个任务
		Future<Integer> result = forkJoinPool.submit(task);
		try {
			System.out.println(result.get());
		} catch (InterruptedException e) {

		} catch (ExecutionException e) {
		}
	}
}

通过这个例子， 可以了解到 ForkJoinTask， ForkJoinTask 与一般任务 的 主要区别在于它需要实现 compute 方法，在这个方法里， 首先需要判断任务是否足够小， 如果 足够小就直接执行任务。 如果不足够小， 就必须分割成 两个子任务， 每个子任务 在 调用 fork 方法时，又会进入 compute 方法， 看看当前子 任务 是否需要继续分割成 子 任务， 如果不需要继续分割， 则执行当前子任务并返回 结果。 使用 join 方法会等待子任务 执行完并得到其结果。

#Fork/Join 的异常处理

ForkJoinTask 在执行的时候可能 会抛出异常，但是我们没办法在主线程里直接捕获异常，所以 ForkJoinTask 提供了 isCompletedAbnormally() 方法来检查任务是否 已经抛出异常或已经被取消了， 并且可以通过 ForkJoinTask 的 getException 方法获取异常。 使用如下代码。

if( task.isCompletedAbnormally()) { 
    System.out.println(task.getException()); 
}

getException 方法返回 Throwable 对象，如果任务被取消了则返回 CancellationException。如果任务没有完成或者没有抛出异常则 返回 null。

#Fork/Join 框架实现原理简单分析

ForkJoinPool 由 ForkJoinTask 数组和 ForkJoinWorkerThread 数组组成， ForkJoinTask 数组 负责将存放程序提交给 ForkJoinPool 的任务，而 ForkJoinWorkerThread 数组负责执行这些任务。

Fork 方法的实现原理

当我们调用 ForkJoinTask 的 fork 方法时，程序会调用 ForkJoinWorkerThread 的 pushTask 方法异步地执行这个任务，然后立即返回结果。

public final ForkJoinTask< V> fork() {
    ((ForkJoinWorkerThread)Thread.currentThread()).pushTask(this); 
    return this; 
    
}

pushTask 方法把当前任务存放在 ForkJoinTask 数组队列里。然后再调用 ForkJoinPool 的 signalWork() 方法唤醒或创建一个工作线程来执行任务。

Join 方法的实现原理

Join 方法的主要作用是阻塞当前 线程并等待获取结果，代码实现如下：

public final V join() { 
    if (doJoin() != NORMAL)     return reportResult(); 
    else return getRawResult(); 
    
} 
private V reportResult() {
    int s; 
    Throwable ex; 
    if ((s = status) == CANCELLED) 
        throw new CancellationException();
    if (s == EXCEPTIONAL && (ex = getThrowableException()) != null) 
        UNSAFE.throwException(ex);
    return getRawResult(); }

它调用了 doJoin() 方法，通过 doJoin() 方法得到 当前任务的状态来判断返回什么结果，根据返回状态来判断是返回任务结果还是直接抛出异常。

++ 总结自《Java 并发编程的艺术》++