ForkJoin框架详解

你的名字 2022-05-28 07:19 158阅读 0赞

# 一、ForkJoin主要类 #

## 主要类有以下4个： ##

1、ForkJoinPool  
2、ForkJoinTask  
3、ForkJoinWorkerThread  
4、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。

# 二、work-stealing #

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

## 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 过来的话，就不会进行重试（看是否有任务可以执行）。

“空闲的” worker thread 从其它 worker thread 的 workQueue 里取得“未执行”的 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

# 四、demo #

import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.Future;
    import java.util.concurrent.RecursiveTask;
    
    public class Demo extends RecursiveTask<Integer> {
    
        private int begin;
        private int end;
    
        public Demo(int begin, int end) {
            this.begin = begin;
            this.end = end;
        }
    
        @Override
        protected Integer compute() {
            System.out.println(Thread.currentThread().getName() + " ... ");
    
            int sum = 0;
            // 拆分任务
            if (end - begin <= 2) {
                // 计算
                for (int i = begin; i <= end; i++) {
                    sum += i;
                }
            } else {
                // 拆分
    
                Demo d1 = new Demo(begin, (begin + end) / 2);
                Demo d2 = new Demo((begin + end)/2 + 1, end);
    
                // 执行任务
                d1.fork();
                d2.fork();
    
                Integer a = d1.join();
                Integer b = d2.join();
    
                sum = a + b;
            }
    
            return sum;
        }
    
        public static void main(String[] args) throws Exception {
    
            ForkJoinPool pool = new ForkJoinPool();
    
            Future<Integer> future = pool.submit(new Demo(1, 1000000000));
    
            System.out.println("....");
    
            System.out.println("计算的值为：" + future.get());
        }
    
    }

出处：[https://blog.csdn.net/hotdust/article/details/71637047][https_blog.csdn.net_hotdust_article_details_71637047]

[https_blog.csdn.net_hotdust_article_details_71637047]: https://blog.csdn.net/hotdust/article/details/71637047