JDK：Fork-Join框架

Posted 2022-12-14 smile4lee

概述

前面的文章【JDK】：Executor框架提到Executor框架，而Fork-Join框架也与Executor框架相关的多线程并行运算的框架。

Fork-Join框架有自己的适用范围。如果一个应用能被分解成多个子任务，并且组合多个子任务的结果就能够获得最终的答案，那么这个应用就适合用 Fork-Join框架模式来解决。下图给出了一个 Fork-Join框架模式的示意图，位于图上部的 Task 依赖于位于其下的 Task 的执行，只有当所有的子任务都完成之后，调用者才能获得 Task 0 的返回结果。

Fork-Join框架能够解决很多种类的并行问题。软件开发人员只需要关注任务的划分和中间结果的组合就能充分利用并行平台的优良性能。其他和并行相关的诸多难于处理的问题，例如负载平衡、同步等，都可以由框架采用统一的方式解决。

Fork-Join框架是ExecutorService接口的一种具体实现，目的是为了更好地利用多处理器。它是为那些能够被递归地拆解成子任务的工作类型量身设计的。类似于ExecutorService接口的其他实现，Fork-Join框架会将任务分发给线程池中的工作线程。

Fork-Join框架的主要类图和继承关系如下：

Fork-Join框架的核心是ForkJoinPool类，它是对AbstractExecutorService类的扩展。ForkJoinPool实现了工作窃取算法，并可以执行ForkJoinTask任务。

工作窃取算法

Fork-Join框架通过一种称作工作窃取（work stealing） 的技术减少了工作队列的争用情况。每个工作线程都有自己的工作队列，这是使用双端队列（或者叫做 deque）来实现的（Java 6 在类库中添加了几种 deque 实现，包括 ArrayDeque 和 LinkedBlockingDeque）。当一个任务划分一个新线程时，它将自己推到 deque 的头部。当一个任务执行与另一个未完成任务的合并操作时，它会将另一个任务推到队列头部并执行，而不会休眠以等待另一任务完成（像 Thread.join() 的操作一样）。当线程的任务队列为空，它将尝试从另一个线程的 deque 的尾部 窃取另一个任务。

可以使用标准队列实现工作窃取，但是与标准队列相比，deque 具有两方面的优势：减少争用和窃取。因为只有工作线程会访问自身的 deque 的头部，deque 头部永远不会发生争用；因为只有当一个线程空闲时才会访问 deque 的尾部，所以也很少存在线程的 deque 尾部的争用（在 fork-join 框架中结合 deque 实现会使这些访问模式进一步减少协调成本）。跟传统的基于线程池的方法相比，减少争用会大大降低同步成本。此外，这种方法暗含的后进先出（last-in-first-out，LIFO）任务排队机制意味着最大的任务排在队列的尾部，当另一个线程需要窃取任务时，它将得到一个能够分解成多个小任务的任务，从而避免了在未来窃取任务。因此，工作窃取实现了合理的负载平衡，无需进行协调并且将同步成本降到了最小。

核心步骤

Fork-Join框架的核心主要包括分割任务与合并任务结果两个步骤。

第一步分割任务。首先我们需要有一个fork类来把大任务分割成子任务，有可能子任务还是很大，所以还需要不停的分割，直到分割出的子任务足够小。

第二步执行任务并合并结果。分割的子任务分别放在双端队列里，然后几个启动线程分别从双端队列里获取任务执行。子任务执行完的结果都统一放在一个队列里，启动一个线程从队列里拿数据，然后合并这些数据。

Fork-Join使用两个类来完成以上两件事情（参考上面的类图）：

• ForkJoinTask：我们要使用ForkJoin框架，必须首先创建一个ForkJoin任务。它提供在任务中执行fork()和join()操作的机制，通常情况下我们不需要直接继承ForkJoinTask类，而只需要继承它的子类，Fork/Join框架提供了以下两个子类：
  
  • RecursiveAction：用于没有返回结果的任务。
  • RecursiveTask ：用于有返回结果的任务。
• ForkJoinPool ：ForkJoinTask需要通过ForkJoinPool来驱动执行，任务分割出的子任务会添加到当前工作线程所维护的双端队列中，进入队列的头部。当一个工作线程的队列里暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务。

部分源码

ForkJoinPool由ForkJoinTask数组和ForkJoinWorkerThread数组组成，ForkJoinTask数组负责存放程序提交给ForkJoinPool的任务，而ForkJoinWorkerThread数组负责执行这些任务。

ForkJoinTask的fork()

当我们调用ForkJoinTask的fork方法时，程序会调用ForkJoinWorkerThread的pushTask方法异步的执行这个任务，然后立即返回结果。代码如下：

public final ForkJoinTask fork()           
    ((ForkJoinWorkerThread) Thread.currentThread()).pushTask(this);         
    return this; 
 

pushTask方法把当前任务存放在ForkJoinTask 数组queue里。然后再调用ForkJoinPool的signalWork()方法唤醒或创建一个工作线程来执行任务。代码如下：

    final void pushTask(ForkJoinTask t) 
        ForkJoinTask[] q; int s, m;
        if ((q = queue) != null)     // ignore if queue removed
            long u = (((s = queueTop) & (m = q.length - 1)) << ASHIFT) + ABASE;
            UNSAFE.putOrderedObject(q, u, t);
            queueTop = s + 1;         // or use putOrderedInt
            if ((s -= queueBase) <= 2)
                pool.signalWork();
            else if (s == m)
                growQueue();
        
    

ForkJoinTask的join()

Join方法的主要作用是阻塞当前线程并等待获取结果。代码如下：

public final V join() 
        if (doJoin() != NORMAL)
            return reportResult();
        else
            return getRawResult();


private V reportResult() 
        int s; Throwable ex;
        if ((s = status) == CANCELLED)
            throw new CancellationException();
        if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
                    UNSAFE.throwException(ex);
                return getRawResult();
        

首先，它调用了doJoin()方法，通过doJoin()方法得到当前任务的状态来判断返回什么结果，任务状态有四种：
已完成（NORMAL），被取消（CANCELLED），信号（SIGNAL）和出现异常（EXCEPTIONAL）。

• 如果任务状态是已完成，则直接返回任务结果。
• 如果任务状态是被取消，则直接抛出CancellationException。
• 如果任务状态是抛出异常，则直接抛出对应的异常。

doJoin()方法的实现代码：

private int doJoin() 
        Thread t; 
        ForkJoinWorkerThread w; 
        int s; 
        boolean completed;
        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) 
            if ((s = status) < 0)
                return s;
            if ((w = (ForkJoinWorkerThread)t).unpushTask(this)) 
                try 
                    completed = exec();
                 catch (Throwable rex) 
                    return setExceptionalCompletion(rex);
                
                if (completed)
                    return setCompletion(NORMAL);
            
            return w.joinTask(this);
        
        else
            return externalAwaitDone();
    

在doJoin()方法里，首先通过查看任务的状态，看任务是否已经执行完了，如果执行完了，则直接返回任务状态，如果没有执行完，则从任务数组里取出任务并执行。如果任务顺利执行完成了，则设置任务状态为NORMAL，如果出现异常，则纪录异常，并将任务状态设置为EXCEPTIONAL。

参考和推荐阅读

以下的文章中提到了如何使用Fork-Join的算例，例如获取最大值、数字累加、Fibonacci数列、合并排序等等。

1. 方腾飞，聊聊并发（八）——Fork/Join框架介绍
2. JDK 7 中的 Fork/Join 模式
3. Java 理论与实践: 应用 fork-join 框架