王者并发课 - 钻石 2：分而治之 - 如何从原理深入理解 ForkJoinPool 的快与慢

solve(problem):    if problem is small enough:        // 如果任务足够小，执行任务        solve problem directly (sequential algorithm)    else:        // 拆分任务        for part in subdivide(problem)            fork subtask to solve(part)        // 合并结果        join all subtasks spawned in previous loop        return combined results

public class TheKingRecursiveSumTask extends RecursiveTask<Long> {    private static final AtomicInteger taskCount = new AtomicInteger();    private final int sumBegin;    private final int sumEnd;    /**     * 任务拆分阈值，当任务尺寸大于该值时，进行拆分     */    private final int threshold;
    public TheKingRecursiveSumTask(int sumBegin, int sumEnd, int threshold) {        this.sumBegin = sumBegin;        this.sumEnd = sumEnd;        this.threshold = threshold;    }
    @Override    protected Long compute() {        if ((sumEnd - sumBegin) > threshold) {            // 两个数之间的差值大于阈值，拆分任务            TheKingRecursiveSumTask subTask1 = new TheKingRecursiveSumTask(sumBegin, (sumBegin + sumEnd) / 2, threshold);            TheKingRecursiveSumTask subTask2 = new TheKingRecursiveSumTask((sumBegin + sumEnd) / 2, sumEnd, threshold);            subTask1.fork();            subTask2.fork();            taskCount.incrementAndGet();            return subTask1.join() + subTask2.join();        }        // 直接执行结果        long result = 0L;        for (int i = sumBegin; i < sumEnd; i++) {            result += i;        }        return result;    }
    public static AtomicInteger getTaskCount() {        return taskCount;    }}

 public static void main(String[] args) {     int sumBegin = 0, sumEnd = 10000000;     computeByForkJoin(sumBegin, sumEnd);     computeBySingleThread(sumBegin, sumEnd); }
 private static void computeByForkJoin(int sumBegin, int sumEnd) {     ForkJoinPool forkJoinPool = new ForkJoinPool(16);     long forkJoinStartTime = System.nanoTime();     TheKingRecursiveSumTask theKingRecursiveSumTask = new TheKingRecursiveSumTask(sumBegin, sumEnd, 100);     long forkJoinResult = forkJoinPool.invoke(theKingRecursiveSumTask);     System.out.println("======");     System.out.println("ForkJoin任务拆分：" + TheKingRecursiveSumTask.getTaskCount());     System.out.println("ForkJoin计算结果：" + forkJoinResult);     System.out.println("ForkJoin计算耗时：" + (System.nanoTime() - forkJoinStartTime) / 1000000); }
 private static void computeBySingleThread(int sumBegin, int sumEnd) {     long computeResult = 0 L;     long startTime = System.nanoTime();     for (int i = sumBegin; i < sumEnd; i++) {         computeResult += i;     }     System.out.println("======");     System.out.println("单线程计算结果：" + computeResult);     System.out.println("单线程计算耗时：" + (System.nanoTime() - startTime) / 1000000); }

======ForkJoin任务拆分：131071ForkJoin计算结果：49999995000000ForkJoin计算耗时：207======单线程计算结果：49999995000000单线程计算耗时：40
Process finished with exit code 0

 public ForkJoinPool() {        this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),             defaultForkJoinWorkerThreadFactory, null, false); }

 public ForkJoinPool(int parallelism) {        this(parallelism, defaultForkJoinWorkerThreadFactory, null, 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 <T> T invoke(ForkJoinTask<T> task) {        if (task == null)            throw new NullPointerException();        externalPush(task);        return task.join(); }

 public void execute(ForkJoinTask<?> task) {        if (task == null)            throw new NullPointerException();        externalPush(task); }
public void execute(Runnable task) {        if (task == null)            throw new NullPointerException();        ForkJoinTask<?> job;        if (task instanceof ForkJoinTask<?>) // avoid re-wrap            job = (ForkJoinTask<?>) task;        else            job = new ForkJoinTask.RunnableExecuteAction(task);        externalPush(job); }

   public < T > ForkJoinTask < T > submit(ForkJoinTask < T > task) {       if (task == null)           throw new NullPointerException();       externalPush(task);       return task;   }
   public < T > ForkJoinTask < T > submit(Callable < T > task) {       ForkJoinTask < T > job = new ForkJoinTask.AdaptedCallable < T > (task);       externalPush(job);       return job;   }
   public < T > ForkJoinTask < T > submit(Runnable task, T result) {       ForkJoinTask < T > job = new ForkJoinTask.AdaptedRunnable < T > (task, result);       externalPush(job);       return job;   }
   public ForkJoinTask < ? > submit(Runnable task) {       if (task == null)           throw new NullPointerException();       ForkJoinTask < ? > job;       if (task instanceof ForkJoinTask < ? > ) // avoid re-wrap           job = (ForkJoinTask < ? > ) task;       else           job = new ForkJoinTask.AdaptedRunnableAction(task);       externalPush(job);       return job;   }

    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;    // 如果调用doJoin返回的非NORMAL状态，将报告异常    if ((s = doJoin() & DONE_MASK) != NORMAL)        reportException(s);    // 正常执行结束，返回原始结果    return getRawResult();}
private int doJoin() {    int s;    Thread t;    ForkJoinWorkerThread wt;    ForkJoinPool.WorkQueue w;    //如果已完成，返回状态    return (s = status) < 0 ? s :    　//如果未完成且当前线程是ForkJoinWorkerThread，则从该线程中取出workQueue，并尝试将当前task取出执行。如果执行的结果是完成，则返回状态；否则，使用当前线程池awaitJoin方法进行等待        ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?        (w = (wt = (ForkJoinWorkerThread) t).workQueue).    tryUnpush(this) && (s = doExec()) < 0 ? s :        wt.pool.awaitJoin(w, this, 0 L):     //当前线程非ForkJoinWorkerThread，调用externalAwaitDone方法等待        externalAwaitDone();}
final int doExec() {    int s;    boolean completed;    if ((s = status) >= 0) {        try {            completed = exec();        } catch (Throwable rex) {            return setExceptionalCompletion(rex);        }        // 执行完成后，将状态设置为NORMAL        if (completed)            s = setCompletion(NORMAL);    }    return s;}

static class SortTask extends RecursiveAction {    final long[] array;    final int lo, hi;    SortTask(long[] array, int lo, int hi) {        this.array = array;        this.lo = lo;        this.hi = hi;    }    SortTask(long[] array) {        this(array, 0, array.length);    }    // 核心计算方法    protected void compute() {        if (hi - lo < THRESHOLD)            // 直接执行            sortSequentially(lo, hi);        else {            // 拆分任务            int mid = (lo + hi) >>> 1;            invokeAll(new SortTask(array, lo, mid),                new SortTask(array, mid, hi));            merge(lo, mid, hi);        }    }    // implementation details follow:    static final int THRESHOLD = 1000;    void sortSequentially(int lo, int hi) {        Arrays.sort(array, lo, hi);    }    void merge(int lo, int mid, int hi) {        long[] buf = Arrays.copyOfRange(array, lo, mid);        for (int i = 0, j = lo, k = mid; i < buf.length; j++)            array[j] = (k == hi || buf[i] < array[k]) ?            buf[i++] : array[k++];    }}

 class Fibonacci extends RecursiveTask<Integer> {   final int n;   Fibonacci(int n) { this.n = n; }   Integer compute() {     if (n <= 1)       return n;     Fibonacci f1 = new Fibonacci(n - 1);     f1.fork();     Fibonacci f2 = new Fibonacci(n - 2);     return f2.compute() + f1.join();   } }

public int getRunningThreadCount() {    int rc = 0;    WorkQueue[] ws;    WorkQueue w;    if ((ws = workQueues) != null) {        for (int i = 1; i < ws.length; i += 2) {            if ((w = ws[i]) != null && w.isApparentlyUnblocked())                ++rc;        }    }    return rc;}

public boolean isQuiescent() {    return (config & SMASK) + (int)(ctl >> AC_SHIFT) <= 0;}

public boolean isQuiescent() {    return (config & SMASK) + (int)(ctl >> AC_SHIFT) <= 0;}

public long getStealCount() {    AtomicLong sc = stealCounter;    long count = (sc == null) ? 0 L : sc.get();    WorkQueue[] ws;    WorkQueue w;    if ((ws = workQueues) != null) {        for (int i = 1; i < ws.length; i += 2) {            if ((w = ws[i]) != null)                count += w.nsteals;        }    }    return count;}

public long getQueuedTaskCount() {    long count = 0;    WorkQueue[] ws;    WorkQueue w;    if ((ws = workQueues) != null) {        for (int i = 1; i < ws.length; i += 2) {            if ((w = ws[i]) != null)                count += w.queueSize();        }    }    return count;}

public int getQueuedSubmissionCount() {    int count = 0;    WorkQueue[] ws;    WorkQueue w;    if ((ws = workQueues) != null) {        for (int i = 0; i < ws.length; i += 2) {            if ((w = ws[i]) != null)                count += w.queueSize();        }    }    return count;}

======ForkJoin任务拆分：16383ForkJoin计算结果：499999999500000000ForkJoin计算耗时：143======单线程计算结果：499999999500000000单线程计算耗时：410