JUC并行计算框架 Fork/Join 原理图文详解&代码示例

    // runState bits: SHUTDOWN must be negative, others arbitrary powers of two
    private static final int  RSLOCK     = 1;
    private static final int  RSIGNAL    = 1 << 1;
    private static final int  STARTED    = 1 << 2;
    private static final int  STOP       = 1 << 29;
    private static final int  TERMINATED = 1 << 30;
    private static final int  SHUTDOWN   = 1 << 31;

    // Instance fields
    volatile long ctl;                   // main pool control
    volatile int runState;               // lockable status
    final int config;                    // parallelism, mode
    int indexSeed;                       // to generate worker index
    volatile WorkQueue[] workQueues;     // main registry
    final ForkJoinWorkerThreadFactory factory;
    final UncaughtExceptionHandler ueh;  // per-worker UEH
    final String workerNamePrefix;       // to create worker name string
    volatile AtomicLong stealCounter;    // also used as sync monitor

ForkJoinPool commonPool =  ForkJoinPool.commonPool();

    /**
     * Returns the common pool instance. This pool is statically
     * constructed; its run state is unaffected by attempts to {@link
     * #shutdown} or {@link #shutdownNow}. However this pool and any
     * ongoing processing are automatically terminated upon program
     * {@link System#exit}.  Any program that relies on asynchronous
     * task processing to complete before program termination should
     * invoke {@code commonPool().}{@link #awaitQuiescence awaitQuiescence},
     * before exit.
     *
     * @return the common pool instance
     * @since 1.8
     */
    public static ForkJoinPool commonPool() {
        // assert common != null : "static init error";
        return common;
    }

    /**
     * Common (static) pool. Non-null for public use unless a static
     * construction exception, but internal usages null-check on use
     * to paranoically [偏执地] avoid potential initialization circularities
     * as well as to simplify generated code.
     */
    static final ForkJoinPool common;

static {
        // initialize field offsets for CAS etc
        try {
            U = sun.misc.Unsafe.getUnsafe();
            Class<?> k = ForkJoinPool.class;
            CTL = U.objectFieldOffset
                (k.getDeclaredField("ctl"));
            RUNSTATE = U.objectFieldOffset
                (k.getDeclaredField("runState"));
            STEALCOUNTER = U.objectFieldOffset
                (k.getDeclaredField("stealCounter"));
            Class<?> tk = Thread.class;
            PARKBLOCKER = U.objectFieldOffset
                (tk.getDeclaredField("parkBlocker"));
            Class<?> wk = WorkQueue.class;
            QTOP = U.objectFieldOffset
                (wk.getDeclaredField("top"));
            QLOCK = U.objectFieldOffset
                (wk.getDeclaredField("qlock"));
            QSCANSTATE = U.objectFieldOffset
                (wk.getDeclaredField("scanState"));
            QPARKER = U.objectFieldOffset
                (wk.getDeclaredField("parker"));
            QCURRENTSTEAL = U.objectFieldOffset
                (wk.getDeclaredField("currentSteal"));
            QCURRENTJOIN = U.objectFieldOffset
                (wk.getDeclaredField("currentJoin"));
            Class<?> ak = ForkJoinTask[].class;
            ABASE = U.arrayBaseOffset(ak);
            int scale = U.arrayIndexScale(ak);
            if ((scale & (scale - 1)) != 0)
                throw new Error("data type scale not a power of two");
            ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
        } catch (Exception e) {
            throw new Error(e);
        }

        commonMaxSpares = DEFAULT_COMMON_MAX_SPARES;
        defaultForkJoinWorkerThreadFactory =
            new DefaultForkJoinWorkerThreadFactory();
        modifyThreadPermission = new RuntimePermission("modifyThread");

        common = java.security.AccessController.doPrivileged
            (new java.security.PrivilegedAction<ForkJoinPool>() {
                public ForkJoinPool run() { return makeCommonPool(); }});
        int par = common.config & SMASK; // report 1 even if threads disabled
        commonParallelism = par > 0 ? par : 1;
    }

    /**
     * Creates and returns the common pool, respecting user settings
     * specified via system properties.
     */
    private static ForkJoinPool makeCommonPool() {
        int parallelism = -1;
        ForkJoinWorkerThreadFactory factory = null;
        UncaughtExceptionHandler handler = null;
        try {  // ignore exceptions in accessing/parsing properties
            String pp = System.getProperty
                ("java.util.concurrent.ForkJoinPool.common.parallelism");
            String fp = System.getProperty
                ("java.util.concurrent.ForkJoinPool.common.threadFactory");
            String hp = System.getProperty
                ("java.util.concurrent.ForkJoinPool.common.exceptionHandler");
            if (pp != null)
                parallelism = Integer.parseInt(pp);
            if (fp != null)
                factory = ((ForkJoinWorkerThreadFactory)ClassLoader.
                           getSystemClassLoader().loadClass(fp).newInstance());
            if (hp != null)
                handler = ((UncaughtExceptionHandler)ClassLoader.
                           getSystemClassLoader().loadClass(hp).newInstance());
        } catch (Exception ignore) {
        }
        if (factory == null) {
            if (System.getSecurityManager() == null)
                factory = defaultForkJoinWorkerThreadFactory;
            else // use security-managed default
                factory = new InnocuousForkJoinWorkerThreadFactory();
        }
        if (parallelism < 0 && // default 1 less than #cores 处理器核数
            (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
            parallelism = 1;
        if (parallelism > MAX_CAP)
            parallelism = MAX_CAP;
        return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,
                                "ForkJoinPool.commonPool-worker-");
    }

/**
 * A thread managed by a {@link ForkJoinPool}, which executes
 * {@link ForkJoinTask}s.
 * This class is subclassable solely for the sake of adding
 * functionality -- there are no overridable methods dealing with
 * scheduling or execution.  However, you can override initialization
 * and termination methods surrounding the main task processing loop.
 * If you do create such a subclass, you will also need to supply a
 * custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to
 * {@linkplain ForkJoinPool#ForkJoinPool use it} in a {@code ForkJoinPool}.
 *
 * @since 1.7
 * @author Doug Lea
 */
public class ForkJoinWorkerThread extends Thread {
    /*
     * ForkJoinWorkerThreads are managed by ForkJoinPools and perform
     * ForkJoinTasks. For explanation, see the internal documentation
     * of class ForkJoinPool.
     *
     * This class just maintains links to its pool and WorkQueue.  The
     * pool field is set immediately upon construction, but the
     * workQueue field is not set until a call to registerWorker
     * completes. This leads to a visibility race, that is tolerated
     * by requiring that the workQueue field is only accessed by the
     * owning thread.
     *
     * Support for (non-public) subclass InnocuousForkJoinWorkerThread
     * requires that we break quite a lot of encapsulation (via Unsafe)
     * both here and in the subclass to access and set Thread fields.
     */

    final ForkJoinPool pool;                // the pool this thread works in
    final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics
}

static final class WorkQueue {
    ......
    // 队列状态
    volatile int qlock;        // 1: locked, < 0: terminate; else 0
    // 下一个出队元素的索引位（主要是为线程窃取准备的索引位置）
    volatile int base;         // index of next slot for poll
    // 下一个入队元素准备的索引位
    int top;                   // index of next slot for push
    // 队列中使用数组存储元素
    ForkJoinTask<?>[] array;   // the elements (initially unallocated)
    // 队列所属的ForkJoinPool（可能为空）
    // 注意，一个ForkJoinPool中会有多个执行线程，还会有比执行线程更多的（或一样多的）队列
    final ForkJoinPool pool;   // the containing pool (may be null)
    // 这个队列所属的归并计算工作线程。注意，工作队列也可能不属于任何工作线程
    final ForkJoinWorkerThread owner; // owning thread or null if shared
    // 记录当前正在进行join等待的其它任务
    volatile ForkJoinTask<?> currentJoin;  // task being joined in awaitJoin
    // 当前正在偷取的任务
    volatile ForkJoinTask<?> currentSteal; // mainly used by helpStealer
    ......
}

        /**
         * Pushes a task. Call only by owner in unshared queues.  (The
         * shared-queue version is embedded in method externalPush.)
         *
         * @param task the task. Caller must ensure non-null.
         * @throws RejectedExecutionException if array cannot be resized
         */
        final void push(ForkJoinTask<?> task) {
            ForkJoinTask<?>[] a; ForkJoinPool p;
// base: 下一个出队元素的索引位（主要是为线程窃取准备的索引位置）
// top: 下一个入队元素准备的索引位
            int b = base, s = top, n;
            if ((a = array) != null) {    // ignore if queue removed
                int m = a.length - 1;     // fenced write for task visibility
// 在指定的对象 ForkJoinTask<?>[] a 中，指定的内存偏移量 ((m & s) << ASHIFT) + ABASE 的位置，赋予一个新的元素
                U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);
// 将workQueue对象本身中的top标示的位置 + 1
                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();
            }
        }

    /**
     * If the current thread is operating in a ForkJoinPool,
     * unschedules and returns, without executing, the next task
     * queued by the current thread but not yet executed, if one is
     * available, or if not available, a task that was forked by some
     * other thread, if available. Availability may be transient, so a
     * {@code null} result does not necessarily imply quiescence of
     * the pool this task is operating in.  This method is designed
     * primarily to support extensions, and is unlikely to be useful
     * otherwise.
     *
     * @return a task, or {@code null} if none are available
     */
    protected static ForkJoinTask<?> pollTask() {
        Thread t; ForkJoinWorkerThread wt;
        return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
            (wt = (ForkJoinWorkerThread)t).pool.nextTaskFor(wt.workQueue) :
            null;
    }

    /**
     * Gets and removes a local or stolen task for the given worker.
     *
     * @return a task, if available
     */
    final ForkJoinTask<?> nextTaskFor(WorkQueue w) {
        for (ForkJoinTask<?> t;;) {
            WorkQueue q; int b;
            if ((t = w.nextLocalTask()) != null)
                return t;
            if ((q = findNonEmptyStealQueue()) == null)
                return null;
            if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null)
                return t;
        }
    }


        /**
         * Takes next task, if one exists, in order specified by mode.
         */
        final ForkJoinTask<?> nextLocalTask() {
            return (config & FIFO_QUEUE) == 0 ? pop() : poll();
        }

    // Specialized scanning

    /**
     * Returns a (probably) non-empty steal queue, if one is found
     * during a scan, else null.  This method must be retried by
     * caller if, by the time it tries to use the queue, it is empty.
     */
    private WorkQueue findNonEmptyStealQueue() {
        WorkQueue[] ws; int m;  // one-shot version of scan loop
        int r = ThreadLocalRandom.nextSecondarySeed();
        if ((ws = workQueues) != null && (m = ws.length - 1) >= 0) {
            for (int origin = r & m, k = origin, oldSum = 0, checkSum = 0;;) {
                WorkQueue q; int b;
                if ((q = ws[k]) != null) {
                    if ((b = q.base) - q.top < 0)
                        return q;
                    checkSum += b;
                }
                if ((k = (k + 1) & m) == origin) {
                    if (oldSum == (oldSum = checkSum))
                        break;
                    checkSum = 0;
                }
            }
        }
        return null;
    }

package i

import java.util.concurrent.ForkJoinPool
import java.util.concurrent.ForkJoinTask
import java.util.concurrent.RecursiveTask

/**
 * @author: Jack
 * 2020-05-03 16:08
 * 斐波那契数列 计算第20个斐波那契数列
 * 以递归的方法定义：F(1)=1，F(2)=1, F(n)=F(n-1)+F(n-2)（n>=2，n∈N*）
 */


class FibonacciComputation(var n: Int) : RecursiveTask<Long>() {
    override fun compute(): Long {
        return when {
            n <= 1 -> 1L
            n == 2 -> 1L
            else -> {
                val f1 = FibonacciComputation(n - 1)
                val f2 = FibonacciComputation(n - 2)
                ForkJoinTask.invokeAll(f1, f2)
                Thread.sleep(10)
                f1.join() + f2.join()
            }
        }
    }
}

fun main() {

    val n = 10

    run {
        val pool = ForkJoinPool.commonPool()

        val s = System.currentTimeMillis()
        val fn = pool.invoke(FibonacciComputation(n))
        val t = System.currentTimeMillis()

        println(pool)
        println("fn=$fn")
        println("Time=${t - s}ms")
    }

    run {
        val s = System.currentTimeMillis()
        val fn = fib(n)
        val t = System.currentTimeMillis()
        println("fn=$fn")
        println("Time=${t - s}ms")
    }

}


fun fib(n: Int): Long {
    return when {
        n <= 1 -> 1L
        n == 2 -> 1L
        else -> {
            val f1 = fib(n - 1)
            val f2 = fib(n - 2)
            // 为了模拟计算密集型任务,我们在这里sleep 10ms
            Thread.sleep(10)
            f2 + f1
        }
    }
}

java.util.concurrent.ForkJoinPool@4b1210ee[Running, parallelism = 11, size = 11, active = 0, running = 0, steals = 24, tasks = 0, submissions = 0]
fn=55
Time=116ms
fn=55
Time=603ms

硬件概览：

  型号名称： MacBook Pro
  型号标识符：    MacBookPro15,1
  处理器名称：    Intel Core i7
  处理器速度：    2.6 GHz
  处理器数目：    1
  核总数：  6
  L2 缓存（每个核）：   256 KB
  L3 缓存：    12 MB
  超线程技术：    已启用
  内存：   16 GB
  Boot ROM 版本：  220.260.171.0.0 (iBridge: 16.16.5200.0.0,0)
  序列号（系统）：  C02Z43JXLVCF
  硬件 UUID：  FAAEE2DB-8F7C-54B1-A0B7-F286C27EA35F

package i

import java.util.concurrent.ForkJoinPool
import java.util.concurrent.ForkJoinTask
import java.util.concurrent.RecursiveTask
import java.util.stream.LongStream

/**
 * @author: Jack
 * 2020-05-03 16:08
 * 斐波那契数列 计算第20个斐波那契数列
 * 以递归的方法定义：F(1)=1，F(2)=1, F(n)=F(n-1)+F(n-2)（n>=2，n∈N*）
 */


class Calculator(var numbers: LongArray, var start: Int, var end: Int) : RecursiveTask<Long>() {
    override fun compute(): Long {
        // 当需要计算的数字个数小于6时，直接采用for loop方式计算结果
        if (end - start < 6) {
            var sum = 0L
            for (i in start..end) {
                sum += numbers[i]
                Thread.sleep(10)
            }
            return sum
        }

        // 把任务一分为二，递归拆分(注意此处有递归)到底拆分成多少分 需要根据具体情况而
        val middle = (start + end) / 2
        val left = Calculator(numbers, start, middle)
        val right = Calculator(numbers, middle + 1, end)
        ForkJoinTask.invokeAll(left, right)
        // 为了模拟计算密集型任务,我们在这里sleep 10ms
        return left.join() + right.join()
    }
}

fun sum(numbers: LongArray): Long {
    var sum = 0L
    for (i in numbers) {
        sum += i
        // 为了模拟计算密集型任务,我们在这里sleep 10ms
        Thread.sleep(10)
    }
    return sum
}


fun main() {

    val numbers = LongStream.rangeClosed(1, 100).toArray()

    run {
        val calculator = Calculator(numbers, 0, numbers.size - 1)
        val pool = ForkJoinPool.commonPool()

        val s = System.currentTimeMillis()
        val fn = pool.invoke(calculator)
        val t = System.currentTimeMillis()

        println(pool)
        println("fn=$fn")
        println("Time=${t - s}ms")
    }

    run {
        val s = System.currentTimeMillis()
        val fn = sum(numbers)
        val t = System.currentTimeMillis()

        println("fn=$fn")
        println("Time=${t - s}ms")
    }

}

java.util.concurrent.ForkJoinPool@3941a79c[Running, parallelism = 11, size = 11, active = 0, running = 0, steals = 17, tasks = 0, submissions = 0]
fn=5050
Time=137ms
fn=5050
Time=1169ms