ForkJoin框架的一些原理知识点

布满荆棘的人生 2022-06-16 08:58 247阅读 0赞

参考：  
[Java 并发编程笔记：如何使用 ForkJoinPool 以及原理][Java _ ForkJoinPool]  
[作者 Doug Lea 的论文——《A Java Fork/Join Framework》][Doug Lea _A Java Fork_Join Framework]  
[ForkJoinPool的commonPool相关参数配置][ForkJoinPool_commonPool]

## 零、ForkJoin主要类 ##

主要类有以下4个：

*  ForkJoinPool
 *  ForkJoinTask
 *  ForkJoinWorkerThread
 *  ForkJoinPool.WorkQueue

功能如下：

*  ForkJoinPool：  
    1，用来执行Task，或生成新的ForkJoinWorkerThread  
    2，执行 ForkJoinWorkerThread 间的 work-stealing 逻辑。
 *  ForkJoinTask：  
    1，执行具体的2叉分支逻辑  
    2，声明以同步/异步方式进行执行
 *  ForkJoinWorkerThread：  
    1，是 ForkJoinPool 内的 worker thread，执行 ForkJoinTask。  
    2，内部有 ForkJoinPool.WorkQueue，来保存要执行的 ForkJoinTask。
 *  ForkJoinPool.WorkQueue：  
    1，保存要执行的ForkJoinTask。

## 一、ForkJoinTask\#invokeAll执行方式： ##

例如，如果是invokeAll(task, task)方法的话，代码如下：

public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
        int s1, s2;
        t2.fork();
        if ((s1 = t1.doInvoke() & DONE_MASK) != NORMAL)
            t1.reportException(s1);
        if ((s2 = t2.doJoin() & DONE_MASK) != NORMAL)
            t2.reportException(s2);
    }

代码执行顺序如下：

1，先执行 t2 的 fork 方法，把 t2 这个 task 加到 workQueue 里去。  
2，再执行 t1 的 doInvoke 方法，让 t1 同步执行。  
3，当 t1 执行完后，执行 t2 的 doJoin 方法，执行 t2 任务。

注意，这里为什么要把 t2 加到 workQueue 里后，再执行 t1 呢？

> 因为把任务加入 workQueue 里后，别的 worker thread 有就可能把这个任务取走，进行执行。

## 二、work-stealing ##

forkjoin 框架是有 work-steal 机制的，这个机制主要功能是：

> “空闲的” worker thread 从其它 worker thread 的 workQueue 里取得“未执行”的 task 然后执行。

### 1, 具体细节机制如下： ###

*  每个 worker thread 维护自己的 scheduling 队列中的“可运行的”task - 队列是一个双向队列（称作：deques），支持 LIFO 和 FIFO 操作。
 *  在task 中生成的“子task”，会被放进生成它的 task 所在的 worker thread 的 双向队列。
 *  worker thread 处理双向队列中的 task 时，使用的是 LIFO 规则（最后进来的，最先被处理）
 *  当worker thread 的队列时没有任务可执行时，它会随机地偷取别的 worker thread 的 work queue 里的 task，然后执行它。在偷取时，使用的是 FIFO 规则，即偷取别人队列里“最先进入”队列的 task。
 *  当 worker thread 执行时，遇到了一个join操作（例如：newTask.join），它会暂停当前的 task  
    的处理，而来处理这个join操作所要执行的任务内容。直到这个join操作的任务执行完后，才会返回刚才暂停任务，继续执行被暂停任务的其它内容。所有 task 都会在不进行“阻塞”情况下完成。  
    （这里的“阻塞”的意思，个人理解为不是IO操作的那种阻塞，而是在任务调试时，没有具体的“阻塞”处理（例如：ArrayBlockingQueue的那种阻塞），或是没有用“阻塞的方式”进行任务调度）
    
    > 之前以为每次调用 fork 方法，都会生成一个线程，看了源码和进行Debug后才知道：根据构造函数中的parallelism值来，决定是否启动新线程。  
    > 在 fork 方法中，`((ForkJoinWorkerThread)t).workQueue.push(this)`这语句会把任务加到“当前线程的workQueue”里，进行排队。然后调用`signalWork`方法，来看是否还可以启动新线程来处理“未分配任务”。如果可以，就启动新线程处理任务。
 *  当 worker thread 没有要执行的 task 或者偷取任务失败时，就会进行暂时等待处理（通过yield，sleep，或者调整优先度等方式），过一段时间再重试看看有没有任务可以执行。如果所有的 worker thread 都处于闲置状态，  
    等待上层的发送 task 过来的话，就不会进行重试（看是否有任务可以执行）。

### 2, work-stealing的“LIFO和FIFO”处理方式有两点好处： ###

1，减少了取 task 时的竞争。worker thread 在执行自己队列任务时，是使用从尾部取。别人从它的队列里偷取任务时，是从队列头部取。所以减少了取时的竞争。

2，被偷取的任务，一般都是最早入队列的任务。这种任务一般来说，都是非常大的任务，是那种需要进行递归分析的的大任务，而不是那种分解完的小任务。所以，减少了任务偷取的次数。  
（注意：在实现上，worker thread 在执行自己队列任务时，不总是 LIFO 方式，可以通过构造函数修改成 FIFO 方式）

### 三、关于双向队列： ###

双向队列在实现方面的主要挑战是“同步”和“its avoidance（不知道怎么翻译）”。即使JVM优化了同步功能，每次 push 和 pop 时都要获取锁的这种操作，也会变成瓶颈。但是，一些策略的改变，提供了一种解决方案：

*  push 和 pop 操作，只针对本线程内的队列。
 *  “偷取”操作可以很方便地通过一个“偷取锁”，来进行限制（双向锁在情况需要时，也可以使“偷取”操作失效）。因此，在队列两端的同步问题上的控制操作，就会减少。
 *  当双向队列要变成空时，可以对pop 或“偷取”操作进行控制。不然，这两个操作要被担保，可以操作disjoint elements of the array

[Java _ ForkJoinPool]: http://blog.dyngr.com/blog/2016/09/15/java-forkjoinpool-internals/
[Doug Lea _A Java Fork_Join Framework]: http://gee.cs.oswego.edu/dl/papers/fj.pdf
[ForkJoinPool_commonPool]: https://segmentfault.com/a/1190000008470012