Fork/Join框架简介

1 引子

fork/join框架是从java1.7开始提供的一个并行处理任务的框架（本篇博客基于jdk1.8分析），它的基本思路是将一个大任务分解成若干个小任务，并行处理多个小任务，最后再汇总合并这些小任务的结果便可得到原来的大任务结果。

从字面意思来理解fork/join框架，"fork"表示“分叉”，它把大任务分解成多个小任务并行处理，而“join”表示“联合”，它将这些小任务的结果合并在一起，最终得到大任务的结果。比如计算累加1+2+3+...+100000,可分解成100个子任务，每个子任务只对1000个自然数求和，最终合并这100个子任务的结果。

 fork/join框架会用到工作窃取（work-stealing）算法，那么什么是工作窃取算法呢？

工作窃取（work-stealing）算法是指某个线程从其他队列里窃取任务来执行。使用工作窃取算法的主要目的是为了并发提高性能、提高执行效率，减少无效的等待。当我们需要处理一个大任务时，我们会将这个任务分解成多个相对独立的子任务，为了减少线程竞争，我们将这些子任务放入到不同的队列中，分别为每个队列新建一个线程，一个线程消费一个队列。在处理任务过程中，操作系统可能给有某些线程分配的时间片较多或某些线程所处理任务的工作量较小，它们会先一步将自己所对应队列中的任务处理完了，而此时其他线程却还没处理完自己所属的任务。此时执行快的线程可以到其他线程对应队列中去“窃取”任务来处理，这样就避免了无意义的干等。此时就会出现多线程访问同一个队列，为了避免线程竞争，一般会使用双端队列，被窃取任务线程永远从双端队列的头部取出任务执行，而窃取任务的线程永远从双端队列的尾部“窃取”任务执行。

 工作窃取算法的优缺点

优点：充分利用线程进行并行计算，减少了线程间的竞争

缺点：在某些情况下还是存在竞争，比如双端队列里只有一个任务时，并且该算法会消耗了更多的系统资源，如创建多个线程和多个双端队列。

2 使用fork/join框架

1) 步骤

如何使用“fork/join框架”？其实这还是比较简单的，正如其名字一样，先"fork"分解任务，再“join”合并结果。

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

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

fork/join框架要使用两个类来完成上面的两个步骤。

①forkjointask : 使用fork/join框架，我们应当提供一个forkjointask任务，若没有任务,fork/join就无从谈起。我它提供了fork()、join()这两个api进行任务的分解、结果的合并，它有两个主要的直接子类，我们一般继承这个两个子类。forkjointask是实现future的抽象类，之前在futuretask源码完整解读中有过对future的介绍，它表示一个异步任务的结果。

• recursivetask ：表示有返回结果的任务

• recursiveaction : 表示没有结果的任务

  

②forkjoinpool : forkjointask任务需要执行器forkjoinpool来执行。forkjoinpool是executorservice的实现类，它代表一个可执行任务的执行器。

 

 任务分割出的子任务会添加到当前工作线程所维护的双端队列中，进入队列的头部。当一个工作线程的队列里暂时没有任务时，它会随机从其他工作线程的队列的尾部获取一个任务。

2) 用例

实现需求：求出1+2+3+....1000的结果。这里的任务需要有结果，应该选择使用recursivetask任务 。另外我们需要考虑如何分割任务，此时我计划让每个子任务执行100个数的相加，因此将分割的阀值设为100，那么fork/join框架会将这个主任务分割成10个子任务，最终将10个子任务的结果合并在一起。

package juc;

import java.util.concurrent.executionexception;
import java.util.concurrent.forkjoinpool;
import java.util.concurrent.future;
import java.util.concurrent.recursivetask;

public class forkjoindemo {
    public static void main(string[] args) {
        accumulationtask task = new accumulationtask(0, 1000);
        forkjoinpool forkjoinpool = new forkjoinpool();
        future<integer> future = forkjoinpool.submit(task);
        try {
            long start = system.currenttimemillis();
            int r = future.get();
            system.out.println("执行‘0+1+2+3+...+1000'计算用时" + (system.currenttimemillis() - start) + "毫秒,其结果是：" + r);
        } catch (interruptedexception | executionexception e) {
            e.printstacktrace();
        }

    }
    static class accumulationtask extends recursivetask<integer> {
        private static final int fork_threshold = 100;
        private final int start;
        private final int end;
        public accumulationtask(int start, int end) {
            this.start = start;
            this.end = end;
        }
        @override
        protected integer compute() {
            boolean iscontinuefork = (end - start) > fork_threshold;
            int sum = 0;
            if (iscontinuefork) {   //大于阀值需要继续分割任务
                //二分法，分成左右两段
                int m = (start + end) / 2;
                accumulationtask lefttask = new accumulationtask(start, m);
                accumulationtask righttask = new accumulationtask(m + 1, end);
                //fork方法：分别执行左右两段任务(若任务不够小，将递归调用compute)
                lefttask.fork();
                righttask.fork();
                //等待左右两段任务执行完，再获取子任务的结果
                int leftresult = lefttask.join();
                int rightresult = righttask.join();
                sum = leftresult + rightresult;
            } else {//任务足够小了，可以直接计算
                for (int i = this.start; i <= this.end; i++) {
                    sum += i;
                }
            }
            return sum;
        }
    }
}

从上面的示例可看出使用fork/join框架的关键在于实现compute方法 。在compute方法中，我们首先要确定任务是否需要继续分割，如果任务足够小、满足预先设定的阀值就可直接执行任务。如果任务仍然很大，就必须继续分割成两个子任务，每个子任务在调用fork方法时，又会进入compute方法，看看当前子任务是否需要继续分割成子任务，如果不需要继续分割，则执行当前子任务并返回结果，使用join方法会等待子任务执行完并得到其结果。

3 fork/join框架的异常处理

forkjointask任务在执行过程中可能会(在执行自身任务的线程中)抛出异常，我们无法在主线程中直接捕获异常，但forkjointask提供了iscompletedabnormally方法来判定任务是否抛出过异常或任务被取消（iscompletednormally方法返回任务是否正常完成的布尔值）。另外forkjointask还提供了getexception方法来获取异常。

getexception返回throwable对象，如果任务被取消了就返回cancellationexception，如果任务未完成或未抛出异常就返回null 。

  简单地获取异常

if (lefttask.iscompletedabnormally()) {
    system.out.println(lefttask.getexception());
}

   捕获异常修改原来的用例

public class forkjoindemo {
    public static void main(string[] args) {
        accumulationtask task = new accumulationtask(0, 1000);
        if(task.iscompletedabnormally()){
            system.out.println(task.getexception());
        }
        forkjoinpool forkjoinpool = new forkjoinpool();
        future<integer> future = forkjoinpool.submit(task);
        try {
            long start = system.currenttimemillis();
            int r = future.get();
          system.out.println("执行‘0+1+2+3+...+1000'计算用时" + (system.currenttimemillis() - start) + "毫秒,其结果是：" + r);
        } catch (interruptedexception | executionexception e) {
            e.printstacktrace();
        }
    }

    static class accumulationtask extends recursivetask<integer> {
        private static final int fork_threshold = 100;
        private final int start;
        private final int end;

        public accumulationtask(int start, int end) {
            this.start = start;
            this.end = end;
        }

        @override
        protected integer compute() {
            boolean iscontinuefork = (end - start) > fork_threshold;
            int sum = 0;
            if (iscontinuefork) {   //大于阀值进需要继续分割任务
                //二分法，分成左右两段
                int m = (start + end) / 2;
                accumulationtask lefttask = new accumulationtask(start, m);
                accumulationtask righttask = new accumulationtask(m + 1, end);
                //fork方法：分别执行左右两段任务(若任务不够小，将递归调用compute)
                lefttask.fork();
                righttask.fork();
                //等待左右两段任务执行完，再获取子任务的结果
                int leftresult = lefttask.join();
                int rightresult = righttask.join();
                sum = leftresult + rightresult;
            } else {
                //任务足够小了，可以直接计算
                    for (int i = this.start; i <= this.end; i++) {
                        sum += i;
                        if (i == 999) throw new illegalstateexception();
                    }
            }
            return sum;
        }
    }
}

4 fork/join框架的实现原理

forkjoinpool中有一个重要的成员变量workqueues ，它是静态内部类workqueue类型的数组。workqueue类中有一个forkjointask类型数组array和一个forkjoinworkerthread成员变量owner, array数组负责将存放程序提交给forkjoinpool的任务,而owner负责执行当前workqueue中的任务。

static final class workqueue{
    // instance fields
    volatile int scanstate;    // versioned, <0: inactive; odd:scanning
    int stackpred;             // pool stack (ctl) predecessor
    int nsteals;               // number of steals
    int hint;                  // randomization and stealer index hint
    int config;                // pool index and mode
    volatile int qlock;        // 1: locked, < 0: terminate; else 0
    volatile int base;         // index of next slot for poll
    int top;                   // index of next slot for push
    forkjointask<?>[] array;   // the elements (initially unallocated)
    final forkjoinpool pool;   // the containing pool (may be null)
    final forkjoinworkerthread owner; // owning thread or null if shared
    volatile thread parker;    // == owner during call to park; else null
    volatile forkjointask<?> currentjoin;  // task being joined in awaitjoin
    volatile forkjointask<?> currentsteal; // mainly used by helpstealer
    //...省略
  }

1) forkjointask的fork方法实现原理

当我们调用forkjointask的fork方法时，程序首先判断当前线程的类型，若是forkjoinworkerthread线程，它会调用forkjoinworkerthread的pushtask方法务提交到当前线程t的workqueue对应的队列中；若是普通线程，它调用externalpush()方法将任务提交到队列中；最后返回这任务本身

public final forkjointask<v> fork() {
    thread t;
    if ((t = thread.currentthread()) instanceof forkjoinworkerthread)
        ((forkjoinworkerthread)t).workqueue.push(this);
    else
        forkjoinpool.common.externalpush(this);
    return this;
}

push方法是forkjoinpool的静态内部类workqueue的成员方法，它的基本逻辑是： 将当前任务放入到forkjointask类型数组array中，然后调用signalwork执行任务，若数组容量不够还将调用growarray对数组array扩容。

final void push(forkjointask<?> task) {
    forkjointask<?>[] a; forkjoinpool p;
    int b = base, s = top, n;
    if ((a = array) != null) {    // ignore if queue removed
        int m = a.length - 1;     // fenced write for task visibility
        u.putorderedobject(a, ((m & s) << ashift) + abase, task);//将task放入到成员变量forkjointask类型数组array中
        u.putorderedint(this, qtop, s + 1);//更新下次入队位置的索引
        if ((n = s - b) <= 1) {
            if ((p = pool) != null)
                //队列中最多只有一个任务了，可以唤醒一个线程或创建一个新线程来执行任务
                p.signalwork(p.workqueues, this);
        }
        else if (n >= m)//数组array容量不够，需要扩容
            growarray();
    }
}

而forkjoinpool的成员方法externalpush的基本逻辑与上面的push方法有些类似，但也有些不同：先确定workqueue的槽位，再将当前任务放到workqueue的成员变量forkjointask数组array中，再调用signalwork执行任务。若workqueues是空的，将调用externalsubmit来初始化workqueues及相关属性。

final void externalpush(forkjointask<?> task) {
    workqueue[] ws; workqueue q; int m;
    int r = threadlocalrandom.getprobe();//探针值，用于计算q在workqueues中的索引槽位
    int rs = runstate; //运行状态
    if ((ws = workqueues) != null && (m = (ws.length - 1)) >= 0 &&  //workqueues非空，且workqueues可放入任务(长度大于1)
        //与hashmap类似，m&r是用来确定数组的索引（取余,这里的r相当于hashmap中node的hash属性），
        //sqmask=ob1111110,（sqmask十进制为126，而126<64*2）它的作用是只用workqueues的64个槽位，
        //而sqmask的二进制最低位为0，又相当于强制将"m & r'的最低位设为0（二进制最低位为零时表示偶数），
        //因此"m & r & sqmask"的结果是小于等于64的偶数。
        (q = ws[m & r & sqmask]) != null && r != 0 && rs > 0 &&
        u.compareandswapint(q, qlock, 0, 1)) {  //锁定q，这里cas更新成功后，q.qlock为1，其他线程就不能cas更新q.qlock了
        forkjointask<?>[] a; int am, n, s;
        if ((a = q.array) != null &&
            (am = a.length - 1) > (n = (s = q.top) - q.base)) {
            int j = ((am & s) << ashift) + abase;
            u.putorderedobject(a, j, task);//将task放入到成员变量forkjointask类型数组array中
            u.putorderedint(q, qtop, s + 1);//更新下次入队位置的索引
            u.putintvolatile(q, qlock, 0);//无条件更新q.qlock，解除对q的锁定
            if (n <= 1)  //队列中最多只有一个任务了，可以唤醒一个线程或创建一个新线程来执行任务
                signalwork(ws, q);
            return;
        }
        u.compareandswapint(q, qlock, 1, 0);//q.array无法容纳新任务时，也要解除对q的锁定
    }
    //完整版本的externalpush，可处理不常见的情况，并在向forkjoinpoll中首次提交第一个任务时执行辅助初始化。 
    //它还会检测外部线程的首次提交，如果索引处的workqueue为空或存在线程竞争，则会创建一个新的共享队列。
    externalsubmit(task);// workqueues是空的，需要初始化workqueues及相关属性，并提交任务
}

2) forkjointask的join方法实现原理

join方法的主要作用是阻塞当前线程并等待获取结果。让我们一起看看forkjointask的join方法的实现:

首先调用dojoin方法来执行并获取任务运行状态，若是非正常完成，就报告异常，若正常完成就返回结果。

public final v join() {
    int s;
    if ((s = dojoin() & done_mask) != normal)
        reportexception(s);
    return getrawresult();
}

forkjointask使用成员变量state表示任务状态，它可能有四种状态,已完成（normal）、被取消（cancelled）、等待信号（signal）和出现异常（exceptional）。done_mask是表示任务状态的位(用来取state二进制形式的最高4位)，smask任务的标签数的掩码(可调用compareandsetforkjointasktag将state置为1)。

    static final int done_mask   = 0xf0000000;  // mask out non-completion bits 
    static final int normal      = 0xf0000000;  // must be negative
    static final int cancelled   = 0xc0000000;  // must be < normal
    static final int exceptional = 0x80000000;  // must be < cancelled
    static final int signal      = 0x00010000;  // must be >= 1 << 16
    static final int smask       = 0x0000ffff;  // short bits for tags

若任务状态是已完成，则可直接返回其结果；若任务被取消，则抛出cancellationexception异常；若执行任务过程中抛出过异常，则直接抛出该异常。

private void reportexception(int s) {
    if (s == cancelled)
        throw new cancellationexception();
    if (s == exceptional)
        rethrow(getthrowableexception());
}
public final void getrawresult() { return null; }//recursiveaction不需要结果，返回null
public final v getrawresult() {//recursiveaction
    //recursiveaction在执行exec方法时主要执行"result = compute();"代码,将计算结果赋值给成员result
    return result;
}

我们再来看看dojoin()方法是怎么做的。dojoin()方法使用了过多的三元运算符，不太容易理解，下面我将三元运算替换成if-else。将方法“翻译”后可以很容易地看出dojoin方法的主要逻辑：首先需要查看任务的状态，若任务已完成(可能是任务取消或抛出异常等非正常完成)，则直接返回任务状态；若任务还没执行完，则从工作队列中取出并执行此任务。若此任务能立刻执行完成（可能是任务取消或抛出异常等非正常完成）就返回此状态，反之就调用forkjoinpool.awaitjoin等待任务执行完成。

    private int dojoin() {
        int s; thread t; forkjoinworkerthread wt; forkjoinpool.workqueue w;
        return (s = status) < 0 ? s :
                ((t = thread.currentthread()) instanceof forkjoinworkerthread) ?
                        (w = (wt = (forkjoinworkerthread)t).workqueue).
                                tryunpush(this) && (s = doexec()) < 0 ? s :
                                wt.pool.awaitjoin(w, this, 0l) :
                        externalawaitdone();
    }
    //翻译后的dojoin方法
    private int dojoin() {
        int s;thread t; forkjoinworkerthread wt; forkjoinpool.workqueue w;
        if ((s = status) < 0) {//任务已完成，直接返回state
            return s;
        } else {
            if ((t = thread.currentthread()) instanceof forkjoinworkerthread) { //当前线程是forkjoinworkerthread线程时
                wt = (forkjoinworkerthread) t;
                w = wt.workqueue;
                //tryunpush取出这个任务
                //doexec准备执行exec方法(exec又调用compute方法),并（若完成）记录状态，但是doexce方法不会等待任务执行完成
                if (w.tryunpush(this) && (s = doexec()) < 0){ 
                    return s;
                }else{
                    return  wt.pool.awaitjoin(w, this, 0l)//等待任务执行完成
                }
            }else{
                return externalawaitdone();//当前线程是普通线程，调用externalawaitdone阻塞当前线程，等待任务完成
            }
        }
    }

参考：《java并发编程的艺术》方腾飞