ForkJoin框架之ForkJoinTask

摘要：前言在前面的文章和响应式编程中提到了和后者毫无疑问是一个线程池前者则是一个类似经典定义的概念官方有一个非常无语的解释就是运行在的一个任务抽象就是运行的线程池框架包含和若干的子类它的核心在于分治和工作窍取最大程度利用线程池中的工作线程避免忙的

在前面的文章"CompletableFuture和响应式编程"中提到了ForkJoinTask和ForkJoinPool,后者毫无疑问是一个线程池,前者则是一个类似FutureTask经典定义的概念.

官方有一个非常无语的解释:ForkJoinTask就是运行在ForkJoinPool的一个任务抽象,ForkJoinPool就是运行ForkJoinTask的线程池.

ForkJoin框架包含ForkJoinTask,ForkJoinWorkerThread,ForkJoinPool和若干ForkJoinTask的子类,它的核心在于分治和工作窍取,最大程度利用线程池中的工作线程,避免忙的忙死,饿的饿死.

ForkJoinTask可以理解为类线程但比线程轻量的实体,在ForkJoinPool中运行的少量ForkJoinWorkerThread可以持有大量的ForkJoinTask和它的子任务.ForkJoinTask同时也是一个轻量的Future,使用时应避免较长阻塞和io.

ForkJoinTask在JAVA8中应用广泛,但它是一个抽象类,它的子类派生了各种用途,如后续计划多带带介绍的CountedCompleter,以及若干JAVA8中stream api定义的与并行流有关的各种操作(ops).

首先看ForkJoinTask的签名.

public abstract class ForkJoinTask implements Future, Serializable 

从签名上看,ForkJoinTask实现了future,也可以序列化,但它不是一个Runnable或Callable.

ForkJoinTask虽然可以序列化,但它只对运行前和后敏感,对于执行过程中不敏感.

先来看task的运行字段:

//volatie修饰的任务状态值,由ForkJoinPool或工作线程修改.
volatile int status; 
static final int DONE_MASK   = 0xf0000000;//用于屏蔽完成状态位. 
static final int NORMAL      = 0xf0000000;//表示正常完成,是负值.
static final int CANCELLED   = 0xc0000000;//表示被取消,负值,且小于NORMAL
static final int EXCEPTIONAL = 0x80000000;//异常完成,负值,且小于CANCELLED
static final int SIGNAL      = 0x00010000;//用于signal,必须不小于1<<16,默认为1<<16.
static final int SMASK       = 0x0000ffff;//后十六位的task标签

很显然,DONE_MASK能够过滤掉所有非NORMAL,非CANCELLED,非EXCEPTIONAL的状态,字段的含义也很直白,后面的SIGNAL和SMASK还不明确,后面再看.

//标记当前task的completion状态,同时根据情况唤醒等待该task的线程.
private int setCompletion(int completion) {
    for (int s;;) {
        //开启一个循环,如果当前task的status已经是各种完成(小于0),则直接返回status,这个status可能是某一次循环前被其他线程完成.
        if ((s = status) < 0)
            return s;
        //尝试将原来的status设置为它与completion按位或的结果.
        if (U.compareAndSwapInt(this, STATUS, s, s | completion)) {
            if ((s >>> 16) != 0)
                //此处体现了SIGNAL的标记作用,很明显,只要task完成(包含取消或异常),或completion传入的值不小于1<<16,
                //就可以起到唤醒其他线程的作用.
                synchronized (this) { notifyAll(); }
            //cas成功,返回参数中的completion.
            return completion;
        }
    }
}

前面用注释解释了这个方法的逻辑,显然该方法是阻塞的,如果传入的参数不能将status设置为负值会如何?

显然,可能会有至多一次的成功cas,并且若满足唤醒的条件,会尝试去唤醒线程,甚至可能因为为了唤醒其他线程而被阻塞在synchonized代码块外;也可能没有一次成功的cas,直到其他线程成功将status置为完成.

//final修饰,运行ForkJoinTask的核心方法.
final int doExec() {
    int s; boolean completed;
    //仅未完成的任务会运行,其他情况会忽略.
    if ((s = status) >= 0) {
        try {
            //调用exec
            completed = exec();
        } catch (Throwable rex) {
            //发生异常,用setExceptionalCompletion设置结果
            return setExceptionalCompletion(rex);
        }
        if (completed)
            //正常完成,调用前面说过的setCompletion,参数为normal,并将返回值作为结果s.
            s = setCompletion(NORMAL);
    }
    //返回s
    return s;
}

//记录异常并且在符合条件时传播异常行为
private int setExceptionalCompletion(Throwable ex) {
    //首先记录异常信息到结果
    int s = recordExceptionalCompletion(ex);
    if ((s & DONE_MASK) == EXCEPTIONAL)
        //status去除非完成态标志位(只保留前4位),等于EXCEPTIONAL.内部传播异常
        internalPropagateException(ex);
    return s;
}
//internalPropagateException方法是一个空方法,留给子类实现,可用于completer之间的异常传递
void internalPropagateException(Throwable ex) {}
//记录异常完成
final int recordExceptionalCompletion(Throwable ex) {
    int s;
    if ((s = status) >= 0) {
        //只能是异常态的status可以记录.
        //hash值禁止重写,不使用子类的hashcode函数.
        int h = System.identityHashCode(this);
        final ReentrantLock lock = exceptionTableLock;
        //异常锁,加锁
        lock.lock();
        try {
            //抹除脏异常,后面叙述
            expungeStaleExceptions();
            //异常表数组.ExceptionNode后面叙述.
            ExceptionNode[] t = exceptionTable;//exceptionTable是一个全局的静态常量,后面叙述
            //用hash值和数组长度进行与运算求一个初始的索引
            int i = h & (t.length - 1);
            for (ExceptionNode e = t[i]; ; e = e.next) {
                //找到空的索引位,就创建一个新的ExceptionNode,保存this,异常对象并退出循环
                if (e == null) {
                    t[i] = new ExceptionNode(this, ex, t[i]);//(1)
                    break;
                }
                if (e.get() == this) //已设置在相同的索引位置的链表中,退出循环.//2
                    break;
            //否则e指向t[i]的next,进入下个循环,直到发现判断包装this这个ForkJoinTask的ExceptionNode已经出现在t[i]这个链表并break(2),
            //或者直到e是null,意味着t[i]出发开始的链表并无包装this的ExceptionNode,则将构建一个新的ExceptionNode并置换t[i],
            //将原t[i]置为它的next(1).整个遍历判断和置换过程处在锁中进行.
            }
        } finally {
            lock.unlock();
        }
        //记录成功,将当前task设置为异常完成.
        s = setCompletion(EXCEPTIONAL);
    }
    return s;
}

//exceptionTable声明
private static final ExceptionNode[] exceptionTable;//全局异常node表
private static final ReentrantLock exceptionTableLock;//上面用到的锁,就是一个普通的可重入锁.
private static final ReferenceQueue