阿里的面试题带你认识ForkJoinPool

我相信大家都用过线程池，比如ExcutorService，比如ThreadPoolExcutor

今天来讲讲ForkJoinPool，它实现于ExcutorService，但又和我们常用的

ThreadPoolExcutor原理不同

前言

随着在硬件上多核处理器的发展和广泛使用，并发编程成为程序员必须掌握的一门技术，在面试中也经常考查面试者并发相关的知识。

今天，我们就从一道阿里的面试题来开始

题目：如何充分利用多核CPU，计算超大数组中所有整数的和？

解析开始

1.单线程相加？

我们最容易想到就是单线程相加，一个for循环搞定。

2.线程池相加？

如果进一步优化，我们会自然而然地想到使用线程池来分段相加，最后再把每个段的结果相加。

3.其它？

Yes，就是我们今天的主角——ForkJoinPool，但是它要怎么实现呢？似乎没怎么用过哈^^

让我们看看上面是那种方法都如何实现

/**
 * 计算1亿个整数的和 */
public class ForkJoinPoolTest01 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 构造数据
        int length = 100000000;
        long[] arr = new long[length];
        for (int i = 0; i < length; i++) {
            arr[i] = ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE);
        }
        // 单线程
        singleThreadSum(arr);
        // ThreadPoolExecutor线程池
        multiThreadSum(arr);
        // ForkJoinPool线程池
        forkJoinSum(arr);

    }

    private static void singleThreadSum(long[] arr) {
        long start = System.currentTimeMillis();

        long sum = 0;
        for (int i = 0; i < arr.length; i++) {
            // 模拟耗时，本文由公从号“彤哥读源码”原创
            sum += (arr[i]/5*5/5*5/5*5/5*5/5*5);
        }

        System.out.println("sum: " + sum);
        System.out.println("single thread elapse: " + (System.currentTimeMillis() - start));

    }

    private static void multiThreadSum(long[] arr) throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();

        int count = 8;
        ExecutorService threadPool = Executors.newFixedThreadPool(count);
        List<Future<Long>> list = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            int num = i;
            // 分段提交任务
            Future<Long> future = threadPool.submit(() -> {
                long sum = 0;
                for (int j = arr.length / count * num; j < (arr.length / count * (num + 1)); j++) {
                    try {
                        // 模拟耗时
                        sum += (arr[j]/5*5/5*5/5*5/5*5/5*5);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                return sum;
            });
            list.add(future);
        }

        // 每个段结果相加
        long sum = 0;
        for (Future<Long> future : list) {
            sum += future.get();
        }

        System.out.println("sum: " + sum);
        System.out.println("multi thread elapse: " + (System.currentTimeMillis() - start));
    }

    private static void forkJoinSum(long[] arr) throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();

        ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
        // 提交任务
        ForkJoinTask<Long> forkJoinTask = forkJoinPool.submit(new SumTask(arr, 0, arr.length));
        // 获取结果
        Long sum = forkJoinTask.get();

        forkJoinPool.shutdown();

        System.out.println("sum: " + sum);
        System.out.println("fork join elapse: " + (System.currentTimeMillis() - start));
    }

    private static class SumTask extends RecursiveTask<Long> {
        private long[] arr;
        private int from;
        private int to;

        public SumTask(long[] arr, int from, int to) {
            this.arr = arr;
            this.from = from;
            this.to = to;
        }

        @Override
        protected Long compute() {
            // 小于1000的时候直接相加，可灵活调整
            if (to - from <= 1000) {
                long sum = 0;
                for (int i = from; i < to; i++) {
                    // 模拟耗时
                    sum += (arr[i]/5*5/5*5/5*5/5*5/5*5);
                }
                return sum;
            }

            // 分成两段任务，本文由公从号“彤哥读源码”原创
            int middle = (from + to) / 2;
            SumTask left = new SumTask(arr, from, middle);
            SumTask right = new SumTask(arr, middle, to);

            // 提交左边的任务
            left.fork();
            // 右边的任务直接利用当前线程计算，节约开销
            Long rightResult = right.compute();
            // 等待左边计算完毕
            Long leftResult = left.join();
            // 返回结果
            return leftResult + rightResult;
        }
    }
}

~~Garnett偷偷地告诉你，实际上计算1亿个整数相加，单线程是最快的，我的电脑大概是100ms左右，使用线程池反而会变慢。~~

~~所以，为了演示ForkJoinPool的牛逼之处，我把每个数都/5*5/5*5/5*5/5*5/5*5了一顿操作，用来模拟计算耗时。~~

来看结果：

sum: 107352457433800662
single thread elapse: 789
sum: 107352457433800662
multi thread elapse: 228
sum: 107352457433800662fork join elapse: 189

可以看到，ForkJoinPool相对普通线程池还是有很大提升的。

什么是ForkJoinPool？

谈到线程池，很多人会想到Executors提供的一些预设的线程池，比如单线程线程池SingleThreadExecutor，固定大小的线程池FixedThreadPool，但是很少有人会注意到其中还提供了一种特殊的线程池：WorkStealingPool，我们点进这个方法，会看到和其他方法不同的是，这种线程池并不是通过ThreadPoolExecutor来创建的，而是ForkJoinPool来创建的

public static ExecutorService newWorkStealingPool() {
        return new ForkJoinPool
            (Runtime.getRuntime().availableProcessors(),
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }

ThreadPoolExecutor应该都很了解了，就是一个基本的存储线程的线程池，需要执行任务的时候就从线程池中拿一个线程来执行。而ForkJoinPool则不仅仅是这么简单，同样也不是ThreadPoolExecutor的代替品，这种线程池是为了实现“分治法”这一思想而创建的，通过把大任务拆分成小任务，然后再把小任务的结果汇总起来就是最终的结果，和MapReduce的思想很类似

最核心的思想可以这样描述：

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

1.fork()

fork()方法类似于线程的Thread.start()方法，但是它不是真的启动一个线程，而是将任务放入到工作队列中。

2.join()

join()方法类似于线程的Thread.join()方法，但是它不是简单地阻塞线程，而是利用工作线程运行其它任务。当一个工作线程中调用了join()方法，它将处理其它任务，直到注意到目标子任务已经完成了。

ForkJoinPool内部原理-工作窃取

work-stealing（工作窃取）算法

ForkJoinPool 的另一个特性是它使用了work-stealing（工作窃取）算法：

线程池内的所有工作线程都尝试找到并执行已经提交的任务，或者是被其他活动任务创建的子任务（如果不存在就阻塞等待）。这种特性使得 ForkJoinPool 在运行多个可以产生子任务的任务，或者是提交的许多小任务时效率更高。尤其是构建异步模型的 ForkJoinPool 时，对不需要合并（join）的事件类型任务也非常适用。

在 ForkJoinPool 中，线程池中每个工作线程（ForkJoinWorkerThread）都对应一个任务队列（WorkQueue），工作线程优先处理来自自身队列的任务（LIFO或FIFO顺序，参数 mode 决定），然后以FIFO的顺序随机窃取其他队列中的任务。

ForkJoinPool中的任务

ForkJoinPool 中的任务分为两种：

一种是本地提交的任务（Submission task，如 execute、submit 提交的任务）；

另外一种是 fork 出的子任务（Worker task）。

两种任务都会存放在 WorkQueue 数组中，但是这两种任务并不会混合在同一个队列里，ForkJoinPool 内部使用了一种随机哈希算法（有点类似 ConcurrentHashMap 的桶随机算法）将工作队列与对应的工作线程关联起来，Submission 任务存放在 WorkQueue 数组的偶数索引位置，Worker 任务存放在奇数索引位。

实质上，Submission 与 Worker 一样，只不过它被限制只能执行它们提交的本地任务，在后面的源码解析中，我们统一称之为“Worker”。

任务的分布情况如下图：

ForkJoinPool原理

初始化ForkJoinPool

ForkJoinPool pool = ForkJoinPool.commonPool()

public static ForkJoinPool commonPool() {
    // assert common != null : "static init error";
    return common;
}

获取ForkJoinPool很简单，直接调用commonPool()。注意，这个方法是jdk1.8才加的，也是推荐的方法，满足大部分场景。

static{
    //...
    common = java.security.AccessController.doPrivileged
            (new java.security.PrivilegedAction<ForkJoinPool>() {
                public ForkJoinPool run() { return makeCommonPool(); }});
    //...
}

private static ForkJoinPool makeCommonPool() {
    //...
    return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,"ForkJoinPool.commonPool-worker-");
}

common在static{}里创建，调用的是makeCommonPool()，最终调用ForkJoinPool的构造函数。

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

parallelism默认是cpu核心数，ForkJoinPool里线程数量依据于它，但不表示最大线程数，不要等同于ThreadPoolExecutor里的corePoolSize或者maximumPoolSize。

factory是线程工程，不是新东西了，默认实现是

DefaultForkJoinWorkerThreadFactory。

workerNamePrefix是其中线程名称的前缀，默认使用“ForkJoinPool-*”

config保存不变的参数，包括了parallelism和mode，供后续读取。mode可选FIFO_QUEUE和LIFO_QUEUE，默认是LIFO_QUEUE，具体用哪种，就要看业务。

ctl是ForkJoinPool中最重要的控制字段，将下面信息按16bit为一组封装在一个long中。

AC: 活动的worker数量；

TC: 总共的worker数量；

SS: WorkQueue状态，第一位表示active的还是inactive，其余十五位表示版本号(对付ABA)；

ID:  这里保存了一个WorkQueue在WorkQueue[]的下标，和其他worker通过字段stackPred组成一个TreiberStack。后文讲的栈顶，指这里下标所在的WorkQueue。

TreiberStack：这个栈的pull和pop使用了CAS，所以支持并发下的无锁操作。

AC和TC初始化时取的是parallelism负数，后续代码可以直接判断正负，为负代表还没有达到目标数量。另外ctl低32位有个技巧可以直接用sp=(int)ctl取得，为负代表存在空闲worker。

线程池缺不了状态的变化，记录字段是runState，具体介绍在后面的“ForkJoinPool状态修改”。

任务ForkJoinTask

ForkJoinPool执行任务的对象是ForkJoinTask，它是一个抽象类，有两个具体实现类RecursiveAction和RecursiveTask。

public abstract class RecursiveAction extends ForkJoinTask<Void> {
    protected abstract void compute();

    public final Void getRawResult() { return null; }

    protected final void setRawResult(Void mustBeNull) { }

    protected final boolean exec() {
        compute();
        return true;
    }
}

public abstract class RecursiveTask<V> extends ForkJoinTask<V> {
    V result;

    protected abstract V compute();

    public final V getRawResult() {
        return result;
    }

    protected final void setRawResult(V value) {
        result = value;
    }

    protected final boolean exec() {
        result = compute();
        return true;
    }
}

ForkJoinTask的抽象方法exec由RecursiveAction和RecursiveTask实现，它被定义为final，具体的执行步骤compute延迟到子类实现。很容易看出RecursiveAction和RecursiveTask的区别，前者没有result，getRawResult返回空，它们对应不需要返回结果和需要返回结果两种场景。

ForkJoinTask里很重要的字段是它的状态status，默认是0，当得出结果时变更为负数，有三种结果：

NORMAL

CANCELLED

EXCEPTIONAL

除此之外，在得出结果之前，任务状态能够被设置为SIGNAL，表示有线程等待这个任务的结果，执行完成后需要notify通知，具体看后文的join。

ForkJoinTask在触发执行后，并不支持其他什么特别操作，只能等待任务执行完成。CountedCompleter是ForkJoinTask的子类，它在子任务协作方面扩展了更多操作。我们聚焦ForkJoinPool主线流程，CountedCompleter相关内容另文再介绍。

WorkQueue

WorkQueue是一个双端队列，它定义在ForkJoinPool类里。

scanState描述WorkQueue当前状态：

偶数表示RUNNING

奇数表示SCANNING

负数表示inactive

stackPred是WorkQueue组成TreiberStack时，保存前者的字段。

ForkJoinPool状态修改

STARTED

STOP

TERMINATED

SHUTDOWN

RSLOCK‍‍‍‍

RSIGNAL

runState记录了ForkJoinPool的运行状态，除了SHUTDOWN是负数，其他都是正数。前面四种不用说了，线程池标准状态流转。在多线程环境修改runState，不能简单想改就改，需要先获取锁，RSLOCK和RSIGNAL就用在这里。

private int lockRunState() {
    int rs;
    return ((((rs = runState) & RSLOCK) != 0 ||
             !U.compareAndSwapInt(this, RUNSTATE, rs, rs |= RSLOCK)) ?
            awaitRunStateLock() : rs);
}

修改前调用lockRunState锁定，检查当前状态，尝试一次使用CAS修改runState为RSLOCK。需要状态变化的机会很少，大多数时间一次就能成功，但不能排除少几率的竞争，这时候进入awaitRunStateLock。

private int awaitRunStateLock() {
    Object lock;
    boolean wasInterrupted = false;
    for (int spins = SPINS, r = 0, rs, ns;;) {
        //1
        if (((rs = runState) & RSLOCK) == 0) {
            if (U.compareAndSwapInt(this, RUNSTATE, rs, ns = rs | RSLOCK)) {
                if (wasInterrupted) {
                    try {
                        Thread.currentThread().interrupt();
                    } catch (SecurityException ignore) {
                    }
                }
                return ns;
            }
        }
        else if (r == 0)
            r = ThreadLocalRandom.nextSecondarySeed();
        else if (spins > 0) {
            r ^= r << 6; r ^= r >>> 21; r ^= r << 7; // xorshift
            if (r >= 0)
                --spins;
        }
        //2
        else if ((rs & STARTED) == 0 || (lock = stealCounter) == null)
            Thread.yield();   // initialization race
        //3
        else if (U.compareAndSwapInt(this, RUNSTATE, rs, rs | RSIGNAL)) {
            synchronized (lock) {
                if ((runState & RSIGNAL) != 0) {
                    try {
                        lock.wait();
                    } catch (InterruptedException ie) {
                        if (!(Thread.currentThread() instanceof
                              ForkJoinWorkerThread))
                            wasInterrupted = true;
                    }
                }
                else
                    lock.notifyAll();
            }
        }
    }
}

在自旋中，第一步，mark1再次尝试修改runState为RSLOCK，成功直接返回。

mark2检查ForkJoinPool初始化情况，这里没有额外多写个变量做锁，直接利用了stealCounter这个原子变量。因为初始化时（后文的externalSubmit），才会对stealCounter赋值。所以当状态不是STARTED或者stealCounter为空时，让出线程等待。

mark3处，线程不会无限制自旋尝试，会利用wait/notify进入阻塞等待。RSIGNAL代替原状态，表示有线程进入了等待，解锁时要处理。在高并发下，这不是一个好的设计，但进入这里的几率很低，作为兜底还是可以的。

private void unlockRunState(int oldRunState, int newRunState) {
    if (!U.compareAndSwapInt(this, RUNSTATE, oldRunState, newRunState)) {
        Object lock = stealCounter;
        runState = newRunState;              // clears RSIGNAL bit
        if (lock != null)
            synchronized (lock) { lock.notifyAll(); }
    }
}

解锁的逻辑就比较简单，如果顺利将状态修改为目标状态，成功大吉。否则表示有别的线程进入了wait，需要调用notifyAll唤醒，重新尝试竞争。

ForkJoinPool最佳实践

（1）最适合的是计算密集型任务

（2）在需要阻塞工作线程时，可以使用ManagedBlocker；

（3）不应该在RecursiveTask的内部使用ForkJoinPool.invoke()/invokeAll()；

总结

（1）ForkJoinPool特别适合于“分而治之”算法的实现；

（2）ForkJoinPool和ThreadPoolExecutor是互补的，不是谁替代谁的关系，二者适用的场景不同；

（3）ForkJoinTask有两个核心方法——fork()和join()，有三个重要子类——RecursiveAction、RecursiveTask和CountedCompleter；

（4）ForkjoinPool内部基于“工作窃取”算法实现；

（5）每个线程有自己的工作队列，它是一个双端队列，自己从队列头存取任务，其它线程从尾部窃取任务；

（6）ForkJoinPool最适合于计算密集型任务，但也可以使用ManagedBlocker以便用于阻塞型任务；

（7）RecursiveTask内部可以少调用一次fork()，利用当前线程处理，这是一种技巧；

Garnett还会不断的分享技术干货的，希望你们是我最好的观众！

乐于输出干货的Java技术公众号：Garnett的Java之路。公众号内有大量的技术文章、海量视频资源、精美脑图，不妨来关注一下！回复【资料】领取大量学习资源和免费书籍！