任务拆分计算利器 ForkJoin 框架玩法详解

long sum = 0;
for (int i = 0; i < array.length; i++) {
    sum += array[i];
}
System.out.println("汇总结果：" + sum);

public class ForkJoinTest {
public static void main(String[] args) throws Exception {
    // 创建2000个数组成的数组
    long[] array = new long[2000];
    // 记录for循环汇总计算的值
    long sourceSum = 0;
    for (int i = 0; i < array.length; i++) {
        array[i] = i;
        sourceSum += array[i];
    }
    System.out.println("for循环汇总计算的值: " + sourceSum);

    System.out.println("---------------");

    // fork/join汇总计算的值
    ForkJoinPool forkJoinPool = new ForkJoinPool();
    ForkJoinTask<Long> taskFuture = forkJoinPool.submit(new SumTask(array, 0, array.length));
    System.out.println("fork/join汇总计算的值: " + taskFuture.get());
}

}

public class SumTask extends RecursiveTask<Long> {

/**
 * 最小任务数组最大容量
 */
private static final int THRESHOLD = 500;

private long[] array;
private int start;
private int end;

public SumTask(long[] array, int start, int end) {
    this.array = array;
    this.start = start;
    this.end = end;
}

@Override
protected Long compute() {
    // 检查任务是否足够小，如果任务足够小,直接计算
    if (end - start &lt;= THRESHOLD) {
        long sum = 0;
        for (int i = start; i &lt; end; i++) {
            sum += this.array[i];
        }
        return sum;
    }
    // 任务太大,一分为二
    int middle = (end + start) / 2;
    // 拆分执行
    SumTask leftTask = new SumTask(this.array, start, middle);
    leftTask.fork();
    SumTask rightTask = new SumTask(this.array, middle, end);
    rightTask.fork();
    System.out.println(&#34;进行任务拆分，leftTask数组区间：&#34; + start + &#34;,&#34; + middle + &#34;；rightTask数组区间：&#34; + middle + &#34;,&#34; + end);
    // 汇总结果
    return leftTask.join() +  rightTask.join();
}

}

for循环汇总计算的值: 1999000
进行任务拆分，leftTask数组区间：0,1000；rightTask数组区间：1000,2000

进行任务拆分，leftTask数组区间：1000,1500；rightTask数组区间：1500,2000

进行任务拆分，leftTask数组区间：0,500；rightTask数组区间：500,1000

fork/join汇总计算的值: 1999000

/**



核心构造方法


@param parallelism   可并行执行的线程数量


@param factory       创建线程的工厂


@param handler       异常捕获处理器


@param asyncMode     任务队列模式，true:先进先出的工作模式，false:先进后出的工作模式

*/

public ForkJoinPool(int parallelism,

ForkJoinWorkerThreadFactory factory,

UncaughtExceptionHandler handler,

boolean asyncMode) {

this(checkParallelism(parallelism),

checkFactory(factory),

handler,

asyncMode ? FIFO_QUEUE : LIFO_QUEUE,

"ForkJoinPool-" + nextPoolId() + "-worker-");

checkPermission();

}

public ForkJoinPool() {

this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),

defaultForkJoinWorkerThreadFactory, null, false);

}

// 创建一个 ForkJoinPool 线程池

ExecutorService forkJoinPool = Executors.newWorkStealingPool();

public class ForkJoinWorkerThread extends Thread {

// 所在的线程池

final ForkJoinPool pool;
// 当前线程下的任务队列

final ForkJoinPool.WorkQueue workQueue;
// 初始化时的构造方法

protected ForkJoinWorkerThread(ForkJoinPool pool) {

// Use a placeholder until a useful name can be set in registerWorker

super("aForkJoinWorkerThread");

this.pool = pool;

this.workQueue = pool.registerWorker(this);

}

}

public abstract class ForkJoinTask<V> implements Future<V>, Serializable {

// 将任务推送到任务队列

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;

}
// 等待任务的执行结果

public final V join() {

int s;

if ((s = doJoin() & DONE_MASK) != NORMAL)

reportException(s);

return getRawResult();

}

}