fork join框架是JDK7中出现的一款高效的工具，Java开发人员可以通过它充分利用现代服务器上的多处理器。它是专门为了那些可以递归划分成许多子模块设计的，目的是将所有可用的处理能力用来提升程序的性能。fork join框架一个巨大的优势是它使用了工作窃取算法，可以完成更多任务的工作线程可以从其它线程中窃取任务来执行。你可以查看这篇文章获得更多信息。
fork join：
- this aticle's author is Doug La
http://gee.cs.oswego.edu/dl/papers/fj.pdf - it could the translation of the passage above.
http://ifeve.com/a-java-fork-join-framework/ - this is a passage write after the poster read the source code of ForkJoinTask
http://ifeve.com/talk-concurrency-forkjoin/
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
public class CountTask extends RecursiveTask {
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 = (Integer)leftTask.join();
int rightResult = (Integer)rightTask.join();
//合并子任务
sum = leftResult + rightResult;
}
return sum;
}
public static void main(String[] args) {
ForkJoinPool forkJoinPool = new ForkJoinPool();
//生成一个计算任务，负责计算1+2+3+4
CountTask task = new CountTask(1, 4);
//执行一个任务
Future result = forkJoinPool.submit(task);
try {
System.out.println(result.get());
} catch (InterruptedException e) {
} catch (ExecutionException e) {
}
}
}
ForkJoinPool forkJoinPool = new ForkJoinPool();
这里会开启线程，这个是处理线程所在的线程池。
ForkJoinPool 的属性{
volatile WorkQueue[] workQueues; // main registry
final ForkJoinWorkerThreadFactory factory;
}
workQueues的属性｛
ForkJoinTask<?>[] array; // the elements (initially unallocated)
final ForkJoinPool pool; /
｝
所以workQueues代表 一些列ForkJoinTask队列。
处理线程来源于：
static final class DefaultForkJoinWorkerThreadFactory
implements ForkJoinWorkerThreadFactory {
public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
return new ForkJoinWorkerThread(pool);
}
}
然后被：
ForkJoinWorkerThread.run() 处理:
public void run() {
if (workQueue.array == null) { // only run once
Throwable exception = null;
try {
onStart();
pool.runWorker(workQueue);
} catch (Throwable ex) {
exception = ex;
} finally {
try {
onTermination(exception);
} catch (Throwable ex) {
if (exception == null)
exception = ex;
} finally {
pool.deregisterWorker(this, exception);
}
}
}
}
然后到这里
pool.runWorker
final void runWorker(WorkQueue w) {
w.growArray(); // allocate queue
int seed = w.hint; // initially holds randomization hint
int r = (seed == 0) ? 1 : seed; // avoid 0 for xorShift
for (ForkJoinTask<?> t;;) {
if ((t = scan(w, r)) != null)
w.runTask(t);
else if (!awaitWork(w, r))
break;
r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
}
}
其中scan 就是传说中的窃取方法。但是没看懂。
所有的一切入口只有两个：
1.threadPool的初始化方法。
2. fork()方法。
