ForkJoinPool大型图文现场（一阅到底 vs 直接收藏）（上）

 /**
 * Abstract base class for tasks that run within a {@link ForkJoinPool}.
 * A {@code ForkJoinTask} is a thread-like entity that is much
 * lighter weight than a normal thread.  Huge numbers of tasks and
 * subtasks may be hosted by a small number of actual threads in a
 * ForkJoinPool, at the price of some usage limitations.
 *
 * @since 1.7
 * @author Doug Lea
 */
public abstract class ForkJoinTask<V> implements Future<V>, Serializable

public abstract class RecursiveAction extends ForkJoinTask<Void>{
    ...
  /**
   * The main computation performed by this task.
   */
  protected abstract void compute();
  ...
}
public abstract class RecursiveTask<V> extends ForkJoinTask<V>{
    ...
  protected abstract void compute();
  ...
}

if(任务小到不用继续拆分）{
    直接计算得到结果
}else{
    拆分子任务
    调用子任务的fork()进行计算
    调用子任务的join()合并计算结果
}

@Slf4j
public class ForkJoinDemo {
    public static void main(String[] args) {
        int n = 20;
        // 为了追踪子线程名称，需要重写 ForkJoinWorkerThreadFactory 的方法
        final ForkJoinPool.ForkJoinWorkerThreadFactory factory = pool -> {
            final ForkJoinWorkerThread worker = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
            worker.setName("my-thread" + worker.getPoolIndex());
            return worker;
        };
        //创建分治任务线程池，可以追踪到线程名称
        ForkJoinPool forkJoinPool = new ForkJoinPool(4, factory, null, false);
        // 快速创建 ForkJoinPool 方法
        // ForkJoinPool forkJoinPool = new ForkJoinPool(4);
        //创建分治任务
        Fibonacci fibonacci = new Fibonacci(n);
        //调用 invoke 方法启动分治任务
        Integer result = forkJoinPool.invoke(fibonacci);
        log.info("Fibonacci {} 的结果是 {}", n, result);
    }
}
@Slf4j
class Fibonacci extends RecursiveTask<Integer> {
    final int n;
    Fibonacci(int n) {
        this.n = n;
    }
    @Override
    public Integer compute() {
        //和递归类似，定义可计算的最小单元
        if (n <= 1) {
            return n;
        }
        // 想查看子线程名称输出的可以打开下面注释
        //log.info(Thread.currentThread().getName());
        Fibonacci f1 = new Fibonacci(n - 1);
        // 拆分成子任务
        f1.fork();
        Fibonacci f2 = new Fibonacci(n - 2);
        // f1.join 等待子任务执行结果
        return f2.compute() + f1.join();
    }
}

volatile int base;         // index of next slot for poll
int top;                   // index of next slot for push

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();
}

static final ForkJoinPool common;
/**
     * @return the common pool instance
     * @since 1.8
     */
public static ForkJoinPool commonPool() {
  // assert common != null : "static init error";
  return common;
}

common = java.security.AccessController.doPrivileged
            (new java.security.PrivilegedAction<ForkJoinPool>() {
                public ForkJoinPool run() { return makeCommonPool(); }});
private static ForkJoinPool makeCommonPool() {
  int parallelism = -1;
  ... 其他默认初始化内容 
    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-");
}

private ForkJoinPool(int parallelism,
                     ForkJoinWorkerThreadFactory factory,
                     UncaughtExceptionHandler handler,
                     int mode,
                     String workerNamePrefix) {
  this.workerNamePrefix = workerNamePrefix;
  this.factory = factory;
  this.ueh = handler;
  this.config = (parallelism & SMASK) | mode;
  long np = (long)(-parallelism); // offset ctl counts
  this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
}    

static final int SMASK        = 0xffff;        // short bits == max index
this.config = (parallelism & SMASK) | mode;

static final int MAX_CAP      = 0x7fff;        // max #workers - 1
private static int checkParallelism(int parallelism) {
        if (parallelism <= 0 || parallelism > MAX_CAP)
            throw new IllegalArgumentException();
        return parallelism;
    }

this.config = parallelism | mode;

 // Mode bits for ForkJoinPool.config and WorkQueue.config
 static final int MODE_MASK    = 0xffff << 16;  // top half of int
 static final int LIFO_QUEUE   = 0;
 static final int FIFO_QUEUE   = 1 << 16;
 static final int SHARED_QUEUE = 1 << 31;       // must be negative

long np = (long)(-parallelism); // offset ctl counts

// Active counts 活跃线程数
private static final int  AC_SHIFT   = 48;
private static final long AC_UNIT    = 0x0001L << AC_SHIFT;
private static final long AC_MASK    = 0xffffL << AC_SHIFT;
// Total counts 总线程数
private static final int  TC_SHIFT   = 32;
private static final long TC_UNIT    = 0x0001L << TC_SHIFT;
private static final long TC_MASK    = 0xffffL << TC_SHIFT;
private static final long ADD_WORKER = 0x0001L << (TC_SHIFT + 15); // sign
// 计算 ctl 
this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);

/*
* Bits and masks for field ctl, packed with 4 16 bit subfields:
* AC: Number of active running workers minus target parallelism
* TC: Number of total workers minus target parallelism
* SS: version count and status of top waiting thread
* ID: poolIndex of top of Treiber stack of waiters
*
* When convenient, we can extract the lower 32 stack top bits
* (including version bits) as sp=(int)ctl.  The offsets of counts
* by the target parallelism and the positionings of fields makes
* it possible to perform the most common checks via sign tests of
* fields: When ac is negative, there are not enough active
* workers, when tc is negative, there are not enough total
* workers.  When sp is non-zero, there are waiting workers.  To
* deal with possibly negative fields, we use casts in and out of
* "short" and/or signed shifts to maintain signedness.
*
* Because it occupies uppermost bits, we can add one active count
* using getAndAddLong of AC_UNIT, rather than CAS, when returning
* from a blocked join.  Other updates entail multiple subfields
* and masking, requiring CAS.
*/