Fork/Join 框架简介
Fork/Join 它可以将一个大的任务拆分成多个子任务进行并行处理,最后将子任务结果合并成最后的计算结果,并进行输出。Fork/Join 框架要完成两件事情:
在 Java 的 Fork/Join 框架中,使用两个类完成上述操作
- ForkJoinTask:我们要使用 Fork/Join 框架,首先需要创建一个 ForkJoin 任务。该类提供了在任务中执行 fork 和 join 的机制。通常情况下我们不需要直接集成 ForkJoinTask 类,只需要继承它的子类,Fork/Join 框架提供了两个子类:
- RecursiveAction:用于没有返回结果的任务
- RecursiveTask:用于有返回结果的任务,继承后可以实现递归(自己调自己)调用的任务
- ForkJoinPool:ForkJoinTask 需要通过 ForkJoinPool 来执行
Fork/Join 框架的实现原理
ForkJoinPool
由ForkJoinTask
数组和ForkJoinWorkerThread
数组组成
ForkJoinTask
数组负责存放以及将程序提交给ForkJoinPool
,而ForkJoinWorkerThread
负责执行这些任务。
fork 方法
fork 方法的实现原理:
当我们调用ForkJoinTask
的fork
方法时,程序会把任务放在ForkJoinWorkerThread
的pushTask
的**workQueue
**中,异步地执行这个任务,然后立即返回结果
1 2 3 4 5 6 7 8
| public final ForkJoinTask<V> fork() { Thread t; if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ((ForkJoinWorkerThread)t).workQueue.push(this); else ForkJoinPool.common.externalPush(this); return this; }
|
pushTask
方法把当前任务存放在ForkJoinTask
数组队列里。然后再调用ForkJoinPool
的signalWork()
方法唤醒或创建一个工作线程来执行任务。代码如下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
| final void push(ForkJoinTask<?> task) { ForkJoinTask<?>[] a; ForkJoinPool p; int b = base, s = top, n; if ((a = array) != null) { int m = a.length - 1; U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task); U.putOrderedInt(this, QTOP, s + 1); if ((n = s - b) <= 1) { if ((p = pool) != null) p.signalWork(p.workQueues, this); } else if (n >= m) growArray(); } }
|
join 方法
Join 方法的主要作用是阻塞当前线程并等待获取结果。让我们一起看看ForkJoinTask
的join
方法的实现,代码如下:
1 2 3 4 5 6
| public final V join() { int s; if ((s = doJoin() & DONE_MASK) != NORMAL) reportException(s); return getRawResult(); }
|
它首先调用 doJoin 方法,通过 doJoin()方法得到当前任务的状态来判断返回什么结果,任务状态有 4 种:
已完成(NORMAL)、被取消(CANCELLED)、信号(SIGNAL)和出现异常(EXCEPTIONAL)
- 如果任务状态是已完成,则直接返回任务结果。
- 如果任务状态是被取消,则直接抛出
CancellationException
- 如果任务状态是抛出异常,则直接抛出对应的异常
让我们分析一下 doJoin 方法的实现
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23
| private int doJoin() { int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w; return (s = status) < 0 ? s : ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? (w = (wt = (ForkJoinWorkerThread)t).workQueue). tryUnpush(this) && (s = doExec()) < 0 ? s : wt.pool.awaitJoin(w, this, 0L) : externalAwaitDone(); }
final int doExec() { int s; boolean completed; if ((s = status) >= 0) { try { completed = exec(); } catch (Throwable rex) { return setExceptionalCompletion(rex); } if (completed) s = setCompletion(NORMAL); } return s; }
|
在 doJoin()方法流程如下:
首先通过查看任务的状态,看任务是否已经执行完成,如果执行完成,则直接返回任务状态
如果没有执行完,则从任务数组里取出任务并执行
如果任务顺利执行完成,则设置任务状态为 NORMAL,如果出现异常,则记录异常,并将任务状态设置为 EXCEPTIONAL
Fork/Join 框架的异常处理
ForkJoinTask 在执行的时候可能会抛出异常,但是我们没办法在主线程里直接捕获异常,所以 ForkJoinTask 提供了isCompletedAbnormally()
方法来检查任务是否已经抛出异常或已经被取消了,并且可以通过ForkJoinTask
的getException()
方法获取异常。(需要自己手动捕获异常!)
getException
()方法返回Throwable
对象,如果任务被取消了则返回CancellationException
。如果任务没有完成或者没有抛出异常则返回null
。
入门案例
场景:生成一个计算任务,计算 1+2+3………+1000,每 100 个数切分一个子任务
这里需要用到返回值 sum,即计算出来的总和,所以继承RecursiveTask
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59
| public class ForkJoinTaskDemo extends RecursiveTask<Integer> {
private int start;
private int end;
private int size = 100;
private int sum;
public ForkJoinTaskDemo(int start, int end) { this.start = start; this.end = end; }
@Override protected Integer compute() { System.out.println("当前计算区间为:" + start + "-----------" + end);
int len = end - start; if (len <= 100) { for (int i = start; i <= end; i++) { sum += i; } } else { int middle = start + size; ForkJoinTaskDemo forkJoinTask1 = new ForkJoinTaskDemo(start, middle); ForkJoinTaskDemo forkJoinTask2 = new ForkJoinTaskDemo(middle + 1, end); forkJoinTask1.fork(); forkJoinTask2.fork(); sum = forkJoinTask1.join() + forkJoinTask2.join(); } return sum; } }
|
写一个测试类进行测试
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
| public class Demo { public static void main(String[] args) { ForkJoinTaskDemo forkJoinTaskDemo = new ForkJoinTaskDemo(1, 1000); ForkJoinPool forkJoinPool = new ForkJoinPool(); ForkJoinTask<Integer> result = forkJoinPool.submit(forkJoinTaskDemo); try { System.out.println(result.get()); } catch (Exception e) { e.printStackTrace(); } finally { forkJoinPool.shutdown(); } } }
|