专栏首页做不甩锅的后端java线程池(五):ForkJoinPool源码分析之一(外部提交及worker执行过程)

java线程池(五):ForkJoinPool源码分析之一(外部提交及worker执行过程)

文章目录

在前文中介绍了如何使用ForkJoinPool和ForkJoin的一些基本原理。现在继续来分析ForkJoin,原本计划从源码开始分析。但是ForkJoinPool的源码太过复杂。后续得分好几部分来讲解。今天先做一个总体的介绍。

1.ForkJoinPool总体介绍

在java中运行ForkJoinPool,经过对源码的分析,实际上,需要4个类来配合运行。这四个类分别是:

  • ForkJoinPool 这是线程池的核心类,也是提供方法供我们使用的入口类。基本上forkJoin的核心操作及线程池的管理方法都由这个类提供。后面将详细介绍。
  • ForkJoinPool.WorkQueue 这是ForkJoinPool类的内部类。也是线程池核心的组成部分。ForkJoinPool线程池将由WorkQueue数组组成。为了进一步提高性能,与ThreadPoolExecutor不一样的是,这没有采用外部传入的任务队列,而是作者自己实现了一个阻塞队列。
  • ForkJoinWorkerThread 线程池中运行的thread也是作者重新定义的。这个类的目的是在于将外部的各种形式的task都转换为统一的ForkJoinTask格式。
  • ForkJoinTask 这是ForkJoinPool支持运行的task抽象类,我们一般使用其子类如RecursiveTask或者RecursiveAction。这个在前文有介绍。这个类提供了任务拆分和结果合并的方法。

2 常量及成员变量

2.1 常量

在ForkJoin中,定义了非常多的常量。

//界限定义
static final int SMASK        = 0xffff;        // short bits == max index
static final int MAX_CAP      = 0x7fff;        // max #workers - 1
static final int EVENMASK     = 0xfffe;        // even short bits
static final int SQMASK       = 0x007e;        // max 64 (even) slots

// Masks and units for WorkQueue.scanState and ctl sp subfield
static final int SCANNING     = 1;             // false when running tasks
static final int INACTIVE     = 1 << 31;       // must be negative
static final int SS_SEQ       = 1 << 16;       // version count

// Mode bits for ForkJoinPool.config and WorkQueue.config
static final int MODE_MASK    = 0xffff << 16;  // top half of int
static final int LIFO_QUEUE   = 0;
static final int FIFO_QUEUE   = 1 << 16;
static final int SHARED_QUEUE = 1 << 31;       // must be negative

上述这些界限用二进制的方式表示如下:

这些常量实际上是用于后续进行二进制位运算的范围常量。

在面还有一部分:

//定义IDLE的超时时间 默认位2秒
private static final long IDLE_TIMEOUT = 2000L * 1000L * 1000L; // 2sec

//idea的容忍度 20ms
private static final long TIMEOUT_SLOP = 20L * 1000L * 1000L;  // 20ms

//静态初始化commonMaxSpares的初始值。这个值远远超出了正常的要求,但也远远低于MAX_CAP和典型的OS线程限制,因此允许jvm在耗尽所需资源之前捕获异常以避免滥用。
private static final int DEFAULT_COMMON_MAX_SPARES = 256;

 //阻塞之前等待自旋的次数,这个默认值是0,以降低对CPU的使用。如果大于零,自旋值必须是2的幂次方,至少是4。值2048导致旋转的时间只有典型上下文切换时间的一小部分。
private static final int SPINS  = 0;

//这个是产生随机性的魔数,用于scan的时候进行计算
private static final int SEED_INCREMENT = 0x9e3779b9;

下面是进行ctl字段控制的一些字段和掩码:

//ctl的计算字段和掩码 
// Lower and upper word masks
private static final long SP_MASK    = 0xffffffffL;
private static final long UC_MASK    = ~SP_MASK;

// Active counts
private static final int  AC_SHIFT   = 48;
private static final long AC_UNIT    = 0x0001L << AC_SHIFT;
private static final long AC_MASK    = 0xffffL << AC_SHIFT;

// Total counts
private static final int  TC_SHIFT   = 32;
private static final long TC_UNIT    = 0x0001L << TC_SHIFT;
private static final long TC_MASK    = 0xffffL << TC_SHIFT;
private static final long ADD_WORKER = 0x0001L << (TC_SHIFT + 15); // sign

上述部分表示位如下二进制:

如下为线程池的运行状态。SHUTDOWN的时候为负数

// runState bits: SHUTDOWN must be negative, others arbitrary powers of two
private static final int  RSLOCK     = 1;
private static final int  RSIGNAL    = 1 << 1;
private static final int  STARTED    = 1 << 2;
private static final int  STOP       = 1 << 29;
private static final int  TERMINATED = 1 << 30;
private static final int  SHUTDOWN   = 1 << 31;

其用二级制的形式表示为如下:

2.2 成员变量

// Instance fields
volatile long ctl;                   // main pool control
volatile int runState;               // lockable status
final int config;                    // parallelism, mode
int indexSeed;                       // to generate worker index
volatile WorkQueue[] workQueues;     // main registry
final ForkJoinWorkerThreadFactory factory;
final UncaughtExceptionHandler ueh;  // per-worker UEH
final String workerNamePrefix;       // to create worker name string
volatile AtomicLong stealCounter;    // also used as sync monitor

上面是最主要的成员变量,用下表来表示:

变量名

类型

说明

ctl

volatile long

线程池的主要控制字段

runState

volatile int

线程池的运行状态,值为常量中对应的值

config

final int

将并行度和mode放到了一个int中,便于后续通过位操作计算

indexSeed

int

随机种子,通过前面的魔数来实现

workQueues

volatile WorkQueue[]

组成workQueue的数组。是线程池的核心数据结构

factory

final ForkJoinWorkerThreadFactory

产生线程的工厂方法

ueh

final UncaughtExceptionHandler

每个worker出现异常之后的处理办法,类似于前面ThreadPoolExecutor的拒绝策略

workerNamePrefix

final String

创建线程名称的前缀

stealCounter

volatile AtomicLong

用于监控steal的计数器

3.构造函数

ForkJoinPool提供了3个构造函数:

3.1 ForkJoinPool(parallelism, factory,handler,asyncMode)

public ForkJoinPool(int parallelism,
                    ForkJoinWorkerThreadFactory factory,
                    UncaughtExceptionHandler handler,
                    boolean asyncMode) {
    this(checkParallelism(parallelism),
         checkFactory(factory),
         handler,
         asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
         "ForkJoinPool-" + nextPoolId() + "-worker-");
    checkPermission();
}

此方法没有进行初始化,但是有些字段需要进行说明:

属性

类型

说明

parallelism

int

表示ForkJoinPool支持的最大并行度

factory

ForkJoinWorkerThreadFactory

用于产生线程的工厂方法

handler

UncaughtExceptionHandler

用于异常处理的拒绝策略

asyncMode

boolean

异步模式,如果为ture,则队列将采用FIFO_QUEUE,实现先进先出,反之则LIFO_QUEUE 实现后进先出

这个方法中会对并行度和factory进程check,以确保系统能支持。 parallelism最大不允许大于MAX_CAP。最小必须大于0。 factory则只是判断是否为空,如果为空则这个地方会出现NPE异常。

3.2 ForkJoinPool(parallelism,factory,handler,mode,workerNamePrefix)

private ForkJoinPool(int parallelism,
                     ForkJoinWorkerThreadFactory factory,
                     UncaughtExceptionHandler handler,
                     int mode,
                     String workerNamePrefix) {
    this.workerNamePrefix = workerNamePrefix;
    this.factory = factory;
    this.ueh = handler;
    this.config = (parallelism & SMASK) | mode;
    long np = (long)(-parallelism); // offset ctl counts
    this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
}

这个方法才是真正的初始化的构造方法。上述计算过程,假定parallelism 为7

然后再计算np和ctl

这样就完成了初始化,需要注意的是,这个位运算操作是ForkJoinPool的核心,需要扎实理解。

3.3 ForkJoinPool()

public ForkJoinPool() {
    this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
         defaultForkJoinWorkerThreadFactory, null, false);
}

无参的构造函数,将通过MAX_CAP和当前系统允许的最大并行度的最小值来指定并行度。使用默认的ThreadFactory。handler位空,默认采用asyncMode为false. 实际实这个方法最终还是调用的上面的两个方法,完成了初始化工作。

4. ForkJoinPool的基本组成

再了解了前面的变量之后,我们可以发现,ForkJoinPool的实际组成是,由一个WorkQueue的数组构成。但是这个数组比较特殊,在偶数位,存放外部调用提交的任务,如通过execute和submit等方法。这种队列称为SubmissionQueue。而另外一种任务是前者在执行过程种通过fork方法产生的新任务。这种队列称为workQueue。SubmissionQueue与WorkQueue的区别在于,WorkQueue的属性“final ForkJoinWorkerThread owner;”是有值的。也就是说,WorkQueue将有ForkJoinWorkerThread线程与之绑定。在运行过程中不断的从WorkQueue中获取任务。如果没有可执行的任务,则将从其他的SubmissionQueue和WorkQueue中窃取任务来执行。前面学习过了工作窃取算法,实际上载WorkQueue上的ForkJoinWorkerThread就是一个窃取者,它从SubmissionQueue中或者去偷的WorkQueue中,按FIFO的方式窃取任务。之后执行。也从自己的WorkQueue中安LIFO或者FIFO的方式执行任务。这取决于模式的设定。默认情况下是采用LIFO的方式在执行。组成如下图所示:

5. ForkJoinPool外部任务的提交

ForkJoinPool在初始化之后,支持三种调用方式 invoke 、execute 和submit。我们来对这三种方式进行分析:

5.1 invoke

public <T> T invoke(ForkJoinTask<T> task) {
    if (task == null)
        throw new NullPointerException();
    externalPush(task);
    return task.join();
}

可以看到invoke是支持有返回值的调用方法,之后调用externalPush进行push。

5.2 execute

public void execute(ForkJoinTask<?> task) {
    if (task == null)
        throw new NullPointerException();
    externalPush(task);
}

而exec方法,则是没有返回值的调用方法。适用于哪些不需要返回结果的计算。此外还有种变体,支持Runnable。实际上是将Runnable转换位ForkJoinTask。

public void execute(Runnable task) {
    if (task == null)
        throw new NullPointerException();
    ForkJoinTask<?> job;
    if (task instanceof ForkJoinTask<?>) // avoid re-wrap
        job = (ForkJoinTask<?>) task;
    else
        job = new ForkJoinTask.RunnableExecuteAction(task);
    externalPush(job);
}

5.3 submit

submit于invoke方法基本一致,唯一的不同是,方法内部没有join方法,需要自行定义。

public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
    if (task == null)
        throw new NullPointerException();
    externalPush(task);
    return task;
}

这样可能更加灵活。 此外也支持Runnable和Callable,而且Runnable也可以支持返回值。 对于Callable,通过AdaptedCallable进行适配。

public <T> ForkJoinTask<T> submit(Callable<T> task) {
    ForkJoinTask<T> job = new ForkJoinTask.AdaptedCallable<T>(task);
    externalPush(job);
    return job;
}

对于Runnable,如果需要返回值,统一还是返回的ForkJoinTask,之后在外层join得到结果。

public <T> ForkJoinTask<T> submit(Runnable task, T result) {
    ForkJoinTask<T> job = new ForkJoinTask.AdaptedRunnable<T>(task, result);
    externalPush(job);
    return job;
}

也可以采用这种方式:

public ForkJoinTask<?> submit(Runnable task) {
    if (task == null)
        throw new NullPointerException();
    ForkJoinTask<?> job;
    if (task instanceof ForkJoinTask<?>) // avoid re-wrap
        job = (ForkJoinTask<?>) task;
    else
        job = new ForkJoinTask.AdaptedRunnableAction(task);
    externalPush(job);
    return job;
}

实际上通过这些提交方法,我们发现,最关键的入口时externalPush方法。下面我们就对这个方法进行分析。

6.外部提交过程分析

现在开始跟踪externalPush的源码,对外部提交过程进行分析。

6.1 externalPush

这是外部提交的唯一入口。

/**
 * 尝试将任务添加到提交者当前的队列中,此方法只处理大多数情况,实际上是根据随机数指定一个workQueues的槽位,如果这个位置存在WorkQueue,则加入队列,然后调用signalWork通知其他工作线程来窃取。反之,则通过externalSubmit处理。这只适用于提交队列存在的普通情况。更复杂的逻辑参考externalSubmit。
 *
 * @param task the task. Caller must ensure non-null.
 */
final void externalPush(ForkJoinTask<?> task) {
    WorkQueue[] ws; WorkQueue q; int m;
    //通过ThreadLocalRandom产生随机数
    int r = ThreadLocalRandom.getProbe();
    //线程池的状态
    int rs = runState;
    //如果ws已经完成初始化,且根据随机数定位的index存在workQueue,且cas的方式加锁成功
    if ((ws = workQueues) != null && (m = (ws.length - 1)) >= 0 &&
        //此处先用随机数和wq的size取&,之后再取SQMASK,这些操作将多余的位的值去除
        (q = ws[m & r & SQMASK]) != null && r != 0 && rs > 0 &&
        //cas的方式加锁 将q中位于QLOCK位置的内存的值如果为0,则改为1,采用cas的方式进行
        U.compareAndSwapInt(q, QLOCK, 0, 1)) {
        ForkJoinTask<?>[] a; int am, n, s;
        //判断q中的task数组是否为空,
        if ((a = q.array) != null &&
            //am为q的长度 这是个固定值,如果这个值大于n n就是目前队列中的元素,实际实这里是判断队列是否有空余的位置
            (am = a.length - 1) > (n = (s = q.top) - q.base)) {
            //j实际上是计算添加到workQueue中的index
            int j = ((am & s) << ASHIFT) + ABASE;
            //将task通过cas的方式插入a的index为j的位置
            U.putOrderedObject(a, j, task);
            //将队列q的QTOP位置的内存加1,实际上就是将TOP增加1
            U.putOrderedInt(q, QTOP, s + 1);
            //以可见的方式将q的QLOCK改为0
            U.putIntVolatile(q, QLOCK, 0);
            //此处,如果队列中的任务小于等于1则通知其他worker来窃取。为什么当任务大于1的时候不通知。而且当没有任务的时候发通知岂不是没有意义?此处不太理解
            if (n <= 1)
                //这是个重点方法,通知其他worker来窃取
                signalWork(ws, q);
            return;
        }
        U.compareAndSwapInt(q, QLOCK, 1, 0);
    }
    externalSubmit(task);
}

实际上externalPush的逻辑只能处理简单的逻辑,对于其他复杂的逻辑,则需要通过externalPush提供,而这些简单的逻辑,实际上就是添加到任务队列。这个任务队列的索引一定是偶数:i = m & r & SQMASK。 这个计算过程,实际上由于SQMASK最后一位为0,因此计算的index的最后一位一定为0,这样导致这个值为偶数。也就是说,workQueues的偶数位存放的是外部提交的任务队列。之后提交成功之后,调用signalWork方法让其他的worker来窃取。

6.2 externalSubmit

这是提交过程中的分支逻辑处理的方法。

/**
 *externalPush的完整版本,处理哪些不常用的逻辑。如第一次push的时候进行初始化、此外如果索引队列为空或者被占用,那么创建一个新的任务队列。
 *
 * @param task the task. Caller must ensure non-null.
 */
private void externalSubmit(ForkJoinTask<?> task) {
   //r是随机数,此处双重检测,确保r不为0
    int r;                                    // initialize caller's probe
    if ((r = ThreadLocalRandom.getProbe()) == 0) {
        ThreadLocalRandom.localInit();
        r = ThreadLocalRandom.getProbe();
    }
    //死循环
    for (;;) {
        WorkQueue[] ws; WorkQueue q; int rs, m, k;
        //move默认为false
        boolean move = false;
        //如果runstate小于0 则线程池处于SHUTDOWN状态,配合进行终止
        if ((rs = runState) < 0) {
            //终止的方法 并抛出异常,拒绝该任务
            tryTerminate(false, false);     // help terminate
            throw new RejectedExecutionException();
        }
        //如果状态不为STARTED 说明此时线程池可用
        else if ((rs & STARTED) == 0 ||     // initialize
                 //如果workQueues为null 或者其length小于1 则说明没用初始化
                 ((ws = workQueues) == null || (m = ws.length - 1) < 0)) {
            int ns = 0;
            //对线程池以CAS的方式加锁,从RUNSTATE变为RSLOCK,如果不为RUNSTATE则自旋
            rs = lockRunState();
            try {
                //如果状态为  RSIGNAL RSLOCK  说明加锁成功
                if ((rs & STARTED) == 0) {
                   //用cas的方式初始化STEALCOUNTER
                    U.compareAndSwapObject(this, STEALCOUNTER, null,
                                           new AtomicLong());
                    // create workQueues array with size a power of two
                    //创建workQueues的数组
                    //根据并行度计算得到config,此处确保p在SMASK范围内,即2个字节
                    int p = config & SMASK; // ensure at least 2 slots
                    //n判断p是否大于1,反之则默认按1处理
                    int n = (p > 1) ? p - 1 : 1;
                    //下列过程是找到大于n的最小的2的幂 这个过程之前在HashMap中演示过
                    n |= n >>> 1; n |= n >>> 2;  n |= n >>> 4;
                    n |= n >>> 8; n |= n >>> 16; n = (n + 1) << 1;
                    //根据并行度计算得到了n,之后根据n确定workQueues的array的大小,这个数组的大小不会超过2^16
                    workQueues = new WorkQueue[n];
                    //将ns的值修改为STARTED
                    ns = STARTED;
                }
            } finally {
                //最后将状态解锁 此时改为STARTED状态,这个计算过程有一点绕 
                unlockRunState(rs, (rs & ~RSLOCK) | ns);
            }
            //实际上这个分支只是创建了外层的workQueues数组,此时数组内的内容还是全部都是空的 
        }
        //如果根据随机数计算出来的槽位不为空,即索引处的队列已经创建,这个地方是外层死循环再次进入的结果
        //需要注意的是这个k的计算过程,SQMASK最低的位为0,这样就导致,无论随机数r怎么变化,得到的结果总是偶数。
        else if ((q = ws[k = r & m & SQMASK]) != null) {
            //如果这个槽位的workQueue未被锁定,则用cas的方式加锁 将其改为1
            if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {
                //拿到这个队列中的array
                ForkJoinTask<?>[] a = q.array;
                //s为top索引
                int s = q.top;
                //初始化submitted状态
                boolean submitted = false; // initial submission or resizing
                try {                      // locked version of push
                    //与上面的externalPush一致,此处push到队列中
                    //先判断 数组不为空且数组中有空余位置,能够容纳这个task
                    if ((a != null && a.length > s + 1 - q.base) ||
                        //或者通过初始化的双端队列的数组不为null 
                        (a = q.growArray()) != null) {
                        //计算数组的index
                        int j = (((a.length - 1) & s) << ASHIFT) + ABASE;
                        //在索引index处插入task
                        U.putOrderedObject(a, j, task);
                        //将队列的QTOP加1
                        U.putOrderedInt(q, QTOP, s + 1);
                        //将提交成功状态改为true
                        submitted = true;
                    }
                } finally {
                    //最终采用cas的方式进行解锁 将队列的锁定状态改为0
                    U.compareAndSwapInt(q, QLOCK, 1, 0);
                }
                //如果submitted为true说明数据添加成功,此时调用其他worker来窃取
                if (submitted) {
                    //调用窃取的方法
                    signalWork(ws, q);
                    //退出
                    return;
                }
            }
            //move状态改为true
            move = true;                   // move on failure
        }
        //如果状态不为RSLOCK 上面两个分支都判断过了,那么此处说明这个索引位置没有初始化
        else if (((rs = runState) & RSLOCK) == 0) { // create new queue
           /new一个新队列
            q = new WorkQueue(this, null);
            //hint 记录随机数
            q.hint = r;
            //计算config SHARED_QUEUE 将确保第一位为1 则这个计算出来的config是负数,这与初始化的方法是一致的
            q.config = k | SHARED_QUEUE;
            //将scan状态改为INCATIVE
            q.scanState = INACTIVE;
            //用cas的方式加锁
            rs = lockRunState();  将创建的workQueue push到workQueues的数组中
            // publish index
            if (rs > 0 &&  (ws = workQueues) != null &&
                k < ws.length && ws[k] == null)
                //赋值
                ws[k] = q;                 // else terminated
            //解锁
            unlockRunState(rs, rs & ~RSLOCK);
        }
        else
            //将move改为true
            move = true;                   // move if busy
        if (move)
            //重新计算r
            r = ThreadLocalRandom.advanceProbe(r);
    }
}

这个地方将产生的无owoner的workQueue放置在索引k的位置,需要注意的是k的计算过程,k= r & m & SQMASK。r是随机数,m是数组的长度,而SQMASK:

最后一位不为1,这就导致不管r如何变化,得到的k最后一位都不为1,这就构造了一个偶数k最后一位为0,k不可能是奇数。

6.3 signalWork

上述两个方法如果提交成功,那么调用signalWork,通知工作线程运行。

/**
 * 此处将激活worker Thread。如果工作线程太少则创建,反之则来进行窃取。
 *
 * @param ws the worker array to use to find signallees
 * @param q a WorkQueue --if non-null, don't retry if now empty
 */
final void signalWork(WorkQueue[] ws, WorkQueue q) {
    long c; int sp, i; WorkQueue v; Thread p;
    //如果ctl为负数  ctl初始化的时候就会为负数 如果小于0  说明有任务需要处理
    while ((c = ctl) < 0L) {                       // too few active
        //c为long,强转int 32位的高位都丢弃,此时如果没有修改过ctl那么低位一定为0 可参考前面ctl的推算过程,所以此处sp 为0 sp为0则说明每月空闲的worker
        if ((sp = (int)c) == 0) {                  // no idle workers
            //还是拿c与ADD_WORKER取& 如果不为0 则说明worker太少,需要新增worker
            if ((c & ADD_WORKER) != 0L)            // too few workers
                //通过tryAddWorker 新增worker
                tryAddWorker(c);
            break;
        }
        //再次缺认ws有没有被初始化 如果没有 退出
        if (ws == null)                            // unstarted/terminated
            break;
        //如果ws的length小于sp的最低位 退出
        if (ws.length <= (i = sp & SMASK))         // terminated
            break;
        //如果index处为空 退出
        if ((v = ws[i]) == null)                   // terminating
            break;
        //将sp的低32位取出
        int vs = (sp + SS_SEQ) & ~INACTIVE;        // next scanState
        //计算用sp减去 scanState
        int d = sp - v.scanState;                  // screen CAS
        long nc = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & v.stackPred);
        //采用cas的方式修改ctl 实际上就是加锁 由于ctl的修改可能会导致while循环退出
        if (d == 0 && U.compareAndSwapLong(this, CTL, c, nc)) {
            v.scanState = vs;                      // activate v
            //如果p被park wait中
            if ((p = v.parker) != null)
                //将worker唤醒 
                U.unpark(p);
            //退出
            break;
        }
        //如果此队列为空或者没有task 也退出
        if (q != null && q.base == q.top)          // no more work
            break;
    }
}

可以看到,实际上signalWork就做了两件事情,第一,判断worker是否充足,如果不够,则创建新的worker。第二,判断worker的状态是否被park了,如果park则用unpark唤醒。这样worker就可以取scan其他队列进行窃取了。

6.4 tryAddWorker

此方法是在worker不足的时候,添加一个worker来执行的具体类。

/**
 * 尝试新增一个worker,然后增加ctl中记录的worker的数量
 *
 * @param c incoming ctl value, with total count negative and no
 * idle workers.  On CAS failure, c is refreshed and retried if
 * this holds (otherwise, a new worker is not needed).
 */
private void tryAddWorker(long c) {
    //传入的c为外层调用方法的ctl add标记为false
    boolean add = false;
    do {
        long nc = ((AC_MASK & (c + AC_UNIT)) |
                   (TC_MASK & (c + TC_UNIT)));
        //如果此时ctl与外层传入的ctl相等 说明没有被修改
        if (ctl == c) {
            int rs, stop;                 // check if terminating
            //用cas的方式加锁
            if ((stop = (rs = lockRunState()) & STOP) == 0)
                //增加ctl的数量,如果成功 add为ture
                add = U.compareAndSwapLong(this, CTL, c, nc);
            //解锁
            unlockRunState(rs, rs & ~RSLOCK);
            //如果stop不为0 则说明线程池停止 退出
            if (stop != 0)
                break;
            //如果前面增加ctl中的数量成功,那么此处开始创建worker
            if (add) {
                createWorker();
                break;
            }
        }
    //这个while循环, 前半部分与ADD_WORKER取并,最终只会保留第48位,这个位置为1,同时c的低32为为0,
    } while (((c = ctl) & ADD_WORKER) != 0L && (int)c == 0);
}

实际上这个类中,只是做了一些准备过程,增加count,以及加锁判断,最终还是通过createWorker来进行。

6.5 createWorker

创建worker的具体方法。

// Creating, registering and deregistering workers

/**
 * 创建并启动一个worker,因为前面已经做了增加count,如果此处出现异常,创建worker不成功,则在deregisterWorker中会判断如果ex不为空,且当前为创建状态的话,会重新进入tryAddWorker方法。
 *
 * @return true if successful
 */
private boolean createWorker() {
    //创建线程的工厂方法
    ForkJoinWorkerThreadFactory fac = factory;
    Throwable ex = null;
    ForkJoinWorkerThread wt = null;
    try {
        //如果工厂方法不为空,则用这个工厂方法创建线程,之后再启动线程,此时newThread将与workQueue绑定
        if (fac != null && (wt = fac.newThread(this)) != null) {
            wt.start();
            return true;
        }
    //如果创建失败,出现了异常 则ex变量有值
    } catch (Throwable rex) {
        ex = rex;
    }
    deregisterWorker(wt, ex);
    return false;
}

此方法只是创建了一个ForkJoinThread,实际上worker还是没有创建。实际上这个创建过程是再newThread(this)中来进行的。

ForkJoinWorkerThread(ForkJoinPool pool, ThreadGroup threadGroup,
                     AccessControlContext acc) {
    super(threadGroup, null, "aForkJoinWorkerThread");
    U.putOrderedObject(this, INHERITEDACCESSCONTROLCONTEXT, acc);
    eraseThreadLocals(); // clear before registering
    this.pool = pool;
    this.workQueue = pool.registerWorker(this);
}

再线程创建成功之后调用registerWorker与之绑定。 如果线程创建失败或者出现异常就要调用deregisterWorker对count进行清理或者解除绑定。

6.6 registerWorker

实际上是这个方法,完成worker的创建和绑定。

/**
 * 创建线程,并建立线程与workQueue的关系,此处只会在workQueues数组的奇数位操作
 *
 * @param wt the worker thread
 * @return the worker's queue
 */
final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
    UncaughtExceptionHandler handler;
    //将线程设置为守护线程
    wt.setDaemon(true);                           // configure thread
    //如果没有handler则抛出异常
    if ((handler = ueh) != null)
        wt.setUncaughtExceptionHandler(handler);
    //创建一个workQueue,此时owoner为当前输入的ForkJoinThread
    WorkQueue w = new WorkQueue(this, wt);
    //定义i为0
    int i = 0;                                    // assign a pool index
    //定义mode 
    int mode = config & MODE_MASK;
    //加锁
    int rs = lockRunState();
    try {
        WorkQueue[] ws; int n;                    // skip if no array
        //如果workQueues存在,且长度大于0
        if ((ws = workQueues) != null && (n = ws.length) > 0) {
            //通过魔数计算
            int s = indexSeed += SEED_INCREMENT;  // unlikely to collide
            //m为n-1,而n为数组的初始化长度,第一次创建的时候,n=16,那么m为15
            int m = n - 1;
            //将s左移然后最后一位补上1,之后与奇数m求并集,那么得到的结果必然是奇数
            i = ((s << 1) | 1) & m;               // odd-numbered indices
            //判断i位置是否为空 如果为空,出现了碰撞,则计算步长向后移动 这个步长一定是偶数
            if (ws[i] != null) {                  // collision
                int probes = 0;                   // step by approx half n
                //最小步长为2  n是数组长度,比为偶数,那么如果n小于等于4,则步长为2,反之,则将n右移,将偶数最后一位的0清除,之后再和EVENMASK求并,这样就相当于将原来的长度缩小2倍,并确保是偶数。之后再加上2。那么假定n为16的话,此处计算的step就为10
                int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;
                //之后再通过while循环,继续判断增加步长之后是否碰撞,如果碰撞,则继续增加步长
                while (ws[i = (i + step) & m] != null) {
                    //如果还是碰撞,且probes增加1之后大于长度n,则会触发扩容,workQueues会扩大2倍 这个probes感觉意义不大
                    if (++probes >= n) {
                        workQueues = ws = Arrays.copyOf(ws, n <<= 1);
                        m = n - 1;
                        //将probes置为0
                        probes = 0;
                    }
                }
            }
            //设置随机数seed
            w.hint = s;                           // use as random seed
            //修改config
            w.config = i | mode;
            //修改scanState为i
            w.scanState = i;                      // publication fence
            //将w设置到i处
            ws[i] = w;
        }
    } finally {
       //cas的方式进行解锁
        unlockRunState(rs, rs & ~RSLOCK);
    }
    //此处设置线程name
    wt.setName(workerNamePrefix.concat(Integer.toString(i >>> 1)));
    return w;
}

再registerWorker中,对工作线程和workQueue进行绑定,并设置到workQueues的奇数索引处。注意这里计算得到奇数索引的算法。 //将s左移然后最后一位补上1,之后与奇数m求并集,那么得到的结果必然是奇数。 i = ((s << 1) | 1) & m; 这个位操作,是我们非常值得注意的地方。worker创建成功,之后再将

6.7 deregisterWorker

如果线程创建没有成功,那么count需要回收。以及进行一些清理工作。

/**
 *此方法的主要目的是在创建worker或者启动worker失败之后的回调方法,此时将之前的ctl中增加的count减去。
 *
 * @param wt the worker thread, or null if construction failed
 * @param ex the exception causing failure, or null if none
 */
final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) {
    WorkQueue w = null;
    //如果workQueue和thread不为空
    if (wt != null && (w = wt.workQueue) != null) {
        WorkQueue[] ws;                           // remove index from array
        //根据config计算index
        int idx = w.config & SMASK;
        //加锁
        int rs = lockRunState();
        如果ws不为空且length大于idx同时idx处的worker就是workerQueue 则将该idx处的worker移除
        if ((ws = workQueues) != null && ws.length > idx && ws[idx] == w)
            ws[idx] = null;
        //解锁 修改rs状态 
        unlockRunState(rs, rs & ~RSLOCK);
    }
    //后续对count减少
    long c;                                       // decrement counts
    //死循环 cas的方式将ctl修改
    do {} while (!U.compareAndSwapLong
                 (this, CTL, c = ctl, ((AC_MASK & (c - AC_UNIT)) |
                                       (TC_MASK & (c - TC_UNIT)) |
                                       (SP_MASK & c))));
     //入果workQueue不为空  将其中的task取消                              
    if (w != null) {
        w.qlock = -1;                             // ensure set
        w.transferStealCount(this);
        w.cancelAll();                            // cancel remaining tasks
    }
    //死循环 如果为停止状态则配合停止
    for (;;) {                                    // possibly replace
        WorkQueue[] ws; int m, sp;
        if (tryTerminate(false, false) || w == null || w.array == null ||
            (runState & STOP) != 0 || (ws = workQueues) == null ||
            (m = ws.length - 1) < 0)              // already terminating
            break;
        if ((sp = (int)(c = ctl)) != 0) {         // wake up replacement
            if (tryRelease(c, ws[sp & m], AC_UNIT))
                break;
        }
        else if (ex != null && (c & ADD_WORKER) != 0L) {
            tryAddWorker(c);                      // create replacement
            break;
        }
        else                                      // don't need replacement
            break;
    }
    //异常处理
    if (ex == null)                               // help clean on way out
        ForkJoinTask.helpExpungeStaleExceptions();
    else                                          // rethrow
        ForkJoinTask.rethrow(ex);
}

至此,我们通过外部线程push之后,将任务提交到submissionQueue队列之后,会根据并行度以及工作线程的需要创建workQueue。唤醒工作线程进行窃取的操作就完成了,外部线程的调用就结束了,会回到main方法中去等待结果。

6.8 外部线程提交过程总结

上述代码调用的过程,我们可以形象的用如下图进行表示: 最开始,workQueues是null状态。在第一次执行的时候,externalSubmit方法中会初始化这个数组。

在这之后,还是在externalSubmit方法的for循环中,完成对任务队列的创建,将任务队列创建在偶数索引处。之后将任务写入这个队列:

此后,任务添加完成,但是没有工作队列进行工作。因此在这之后调用signalWork,通知工作队列开始干活。但是在这个方法执行的过程中,由于工作队列并不存在,没有worker,所以调用tryAddWorker开始创建worker。在createWorker会创建一个worker线程:

但是workQueue还没创建。这是在newthread之后,通过registerWorker创建的,在registerWorker方法中,会在奇数位创建一个workQueue,并将此前创建的线程与之绑定。这样一个worker就成功创建了。

这样就完成了worker创建的全过程。

7.workQueue的工作过程

在workQueue创建完成之后,下一步,这些线程的run方法调用后被启动。之后就进入了worker线程的生命周期了。实际上run方方法如下:

public void run() {
    if (workQueue.array == null) { // only run once
        Throwable exception = null;
        try {
            onStart();
            pool.runWorker(workQueue);
        } catch (Throwable ex) {
            exception = ex;
        } finally {
            try {
                onTermination(exception);
            } catch (Throwable ex) {
                if (exception == null)
                    exception = ex;
            } finally {
                pool.deregisterWorker(this, exception);
            }
        }
    }
}

重点执行的式runWorker,此时一旦出错,可以调用deregisterWorker方法进行清理。下面来看看runWorker的详细过程。

7.1 runWorker

这是worker工作线程的执行方法。通过死循环,不断scan是否有任务,之后窃取这个任务进行执行。

/**
 * 通过调用线程的run方法,此时开始最外层的runWorker
 */
final void runWorker(WorkQueue w) {
   //初始化队列,这个方法会根据任务进行判断是否需要扩容
    w.growArray();                   // allocate queue
    //hint是采用的魔数的方式增加
    int seed = w.hint;               // initially holds randomization hint
    //如果seed为0 则使用1
    int r = (seed == 0) ? 1 : seed;  // avoid 0 for xorShift
    //死循环
    for (ForkJoinTask<?> t;;) {
       //调用scan方法 对经过魔数计算的r 之后开始进行窃取过程 如果能够窃取 则task不为空
        if ((t = scan(w, r)) != null)
            //运行窃取之后的task
            w.runTask(t);
        //反之则当前线程进行等待
        else if (!awaitWork(w, r))
            break;
        r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
    }
}

通过scan方法从其他队列获得任务。

7.2 scan

/**
 * 通过scan方法进行任务窃取,扫描从一个随机位置开始,如果出现竞争则通过魔数继续随机移动,反之则线性移动,直到所有队列上的相同校验连续两次出现为空,则说明没有任何任务可以窃取,因此worker会停止窃取,之后重新扫描,如果找到任务则重新激活,否则返回null,扫描工作应该尽可能少的占用内存,以减少对其他扫描线程的干扰。
 *
 * @param w the worker (via its WorkQueue)
 * @param r a random seed
 * @return a task, or null if none found
 */
private ForkJoinTask<?> scan(WorkQueue w, int r) {
    WorkQueue[] ws; int m;
    //如果workQueues不为空且长度大于1,当前workQueue不为空
    if ((ws = workQueues) != null && (m = ws.length - 1) > 0 && w != null) {
        //ss为扫描状态
        int ss = w.scanState;                     // initially non-negative
        //for循环 这是个死循环  origin将r与m求并,将多余位去除。然后赋值给k
        for (int origin = r & m, k = origin, oldSum = 0, checkSum = 0;;) {
            WorkQueue q; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
            int b, n; long c;
            //如果k处不为空
            if ((q = ws[k]) != null) {
                //如果task大于0
                if ((n = (b = q.base) - q.top) < 0 &&
                    (a = q.array) != null) {      // non-empty
                    //计算i
                    long i = (((a.length - 1) & b) << ASHIFT) + ABASE;
                    //得到i处的task 
                    if ((t = ((ForkJoinTask<?>)
                              U.getObjectVolatile(a, i))) != null &&
                        q.base == b) {
                        //如果扫描状态大于0
                        if (ss >= 0) {
                            //更改a中i的值为空 也就是此处将任务窃取走了
                            if (U.compareAndSwapObject(a, i, t, null)) {
                                //将底部的指针加1
                                q.base = b + 1;
                                //如果n小于-1 则通知工作线程工作
                                if (n < -1)       // signal others
                                    signalWork(ws, q);
                                //将窃取的task返回
                                return t;
                            }
                        }
                        //如果 scan状态小于0 则调用tryRelease方法唤醒哪些wait的worker
                        else if (oldSum == 0 &&   // try to activate
                                 w.scanState < 0)
                            //调用tryRelease方法 后续详细介绍
                            tryRelease(c = ctl, ws[m & (int)c], AC_UNIT);
                    }
                    //如果ss小于0 
                    if (ss < 0)                   // refresh
                        //更改ss
                        ss = w.scanState;
                    r ^= r << 1; r ^= r >>> 3; r ^= r << 10;
                    origin = k = r & m;           // move and rescan
                    oldSum = checkSum = 0;
                    continue;
                }
                checkSum += b;
            }
            //此处判断k,k在此通过+1的方式完成对原有workQueues的遍历
            if ((k = (k + 1) & m) == origin) {    // continue until stable
                if ((ss >= 0 || (ss == (ss = w.scanState))) &&
                    oldSum == (oldSum = checkSum)) {
                    if (ss < 0 || w.qlock < 0)    // already inactive
                        break;
                    int ns = ss | INACTIVE;       // try to inactivate
                    long nc = ((SP_MASK & ns) |
                               (UC_MASK & ((c = ctl) - AC_UNIT)));
                    w.stackPred = (int)c;         // hold prev stack top
                    U.putInt(w, QSCANSTATE, ns);
                    if (U.compareAndSwapLong(this, CTL, c, nc))
                        ss = ns;
                    else
                        w.scanState = ss;         // back out
                }
                checkSum = 0;
            }
        }
    }
    return null;
}

此方法最大的特点就是,根据随机数计算一个k,然后根据k去遍历workQueues,之后看看这个位置是否有数据,如果不为空,则检查checkSum,根据checkSum的状态缺认是否从这个队列中取数据。按之前约定的FIFO或者LIFO取数。这意味着,工作队列对窃取和是否获得本队列中的任务之间并没有优先级,而是根据随机数得到的index,之后对数组进行遍历。

7.3 tryRelease

对于执行完成的worker,则需要进行释放。

/**
 * 如果worker处于空闲worker Stack的workQueue的顶部。则发信号对其进行释放。
 *
 * @param c incoming ctl value
 * @param v if non-null, a worker
 * @param inc the increment to active count (zero when compensating)
 * @return true if successful
 */
private boolean tryRelease(long c, WorkQueue v, long inc) {
    //sp取c的低位,计算vs
    int sp = (int)c, vs = (sp + SS_SEQ) & ~INACTIVE; Thread p;
    //如果v不为空 且v的sancState为sp 
    if (v != null && v.scanState == sp) {          // v is at top of stack
        //计算nc
        long nc = (UC_MASK & (c + inc)) | (SP_MASK & v.stackPred);
        //采用cas的方式 将ctl改为nc
        if (U.compareAndSwapLong(this, CTL, c, nc)) {
            //修改scanState的状态
            v.scanState = vs;
            //如果此时这线程为park状态,则调用unpark
            if ((p = v.parker) != null)
                U.unpark(p);
            return true;
        }
    }
    return false;
}

7.4 awaitWork

如果上述scan不到任务呢?也就是说,scan方法没有拿到task,则会调用awaitWork。将当前的线程进行阻塞。

/**
 *如果不能窃取到任务,那么就将worker阻塞。如果停用导致线程池处于静止状态,则检查是否要关闭,如果这不是唯一的工作线程,则等待给定的持续时间,达到超时时间后,如果ctl没有更改,则将这个worker终止,之后唤醒另外一个其他的worker对这个过程进行重复。
 *
 * @param w the calling worker
 * @param r a random seed (for spins)
 * @return false if the worker should terminate
 */
private boolean awaitWork(WorkQueue w, int r) {
    //如果w不为空切w的qlock小于0 则直接返回false 
    if (w == null || w.qlock < 0)                 // w is terminating
        return false;
    //for循环,这是个死循环,定义pred
    for (int pred = w.stackPred, spins = SPINS, ss;;) {
        //如果ss大于0 则返回
        if ((ss = w.scanState) >= 0)
            break;
        //如果spins大于0 
        else if (spins > 0) {
           //计算r
            r ^= r << 6; r ^= r >>> 21; r ^= r << 7;
            //如果r大于0 且spins为0
            if (r >= 0 && --spins == 0) {         // randomize spins
                WorkQueue v; WorkQueue[] ws; int s, j; AtomicLong sc;
                //如果pred不为0 且ws不为空
                if (pred != 0 && (ws = workQueues) != null &&
                     //j<ws.length
                    (j = pred & SMASK) < ws.length &&
                    //j位置处不为空
                    (v = ws[j]) != null &&        // see if pred parking
                     //并且没有park
                    (v.parker == null || v.scanState >= 0))
                    //继续在for循环中自旋
                    spins = SPINS;                // continue spinning
            }
        }
        //如果qlock小于0  返回false
        else if (w.qlock < 0)                     // recheck after spins
            return false;
        //如果被中断
        else if (!Thread.interrupted()) {
            long c, prevctl, parkTime, deadline;
            int ac = (int)((c = ctl) >> AC_SHIFT) + (config & SMASK);
            //如果线程池停止
            if ((ac <= 0 && tryTerminate(false, false)) ||
                (runState & STOP) != 0)           // pool terminating
                return false;
            //最后的等待 获取不到任务 此时采用超时等待  park的方式进行
            if (ac <= 0 && ss == (int)c) {        // is last waiter
                prevctl = (UC_MASK & (c + AC_UNIT)) | (SP_MASK & pred);
                int t = (short)(c >>> TC_SHIFT);  // shrink excess spares
                if (t > 2 && U.compareAndSwapLong(this, CTL, c, prevctl))
                    return false;                 // else use timed wait
                parkTime = IDLE_TIMEOUT * ((t >= 0) ? 1 : 1 - t);
                deadline = System.nanoTime() + parkTime - TIMEOUT_SLOP;
            }
            else
                prevctl = parkTime = deadline = 0L;
            //获取线程
            Thread wt = Thread.currentThread();
            //设置PARKBLOCKER
            U.putObject(wt, PARKBLOCKER, this);   // emulate LockSupport
            w.parker = wt;
            //调用park方法
            if (w.scanState < 0 && ctl == c)      // recheck before park
                U.park(false, parkTime);
            U.putOrderedObject(w, QPARKER, null);
            U.putObject(wt, PARKBLOCKER, null);
            //如果scanState大于0 退出
            if (w.scanState >= 0)
                break;
            //如果已经达到deadline的时间 则返回false
            if (parkTime != 0L && ctl == c &&
                deadline - System.nanoTime() <= 0L &&
                U.compareAndSwapLong(this, CTL, c, prevctl))
                return false;                     // shrink pool
        }
    }
    return true;
}

上述这些方法,如果返回了false,则会导致外层的阻塞不能阻塞住,在runWorker方法中,会退出循环,这个线程就会退出。只有返回了true,那么这个线程就会一直运行,获得任务,循环。

7.5 workQueue的执行过程总结

当workQueue上的thread启动之后,这个线程就会调用runWorker方法。之后再runWorker方法中有一个死循环,不断的通过scan方法去扫描任务,实际上就是执行窃取过程。如下图所示,这样通过遍历外层workQueues的方式将会从任务队列中窃取任务进行执行。

当执行之后,会通过fork产生新的任务,将这些任务任务添加到工作队列中去。其他线程继续scan,执行。这个过程不断循环,直到任务全部都执行完成,这样就完成了整个计算过程。

8.总结

本文对ForkJoinPool的部分源码进行分析,重点分析了外部提交过程和worker的执行过程。但是还有几个比较关键的地方,fork和join的过程,以及工作队列workQueue的组成都还没涉及到,限于篇幅限制,再后面的博客中接着补充。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 聊聊java中的哪些Map:(八)ConcurrentSkipListMap源码分析

    ConcurrentSkipListMap也是java并发包下面的重要容器,其类的继承结构如下:

    冬天里的懒猫
  • 聊聊java中的哪些Map:(九)TreeMap源码分析

    可以看到,TreeMap继承了AbstractMap,此外实现了NavigableMap、Cloneable和Serializable接口。而Navigable...

    冬天里的懒猫
  • 关于ConcurrentHashMap的key和value不能为null的深层次原因

    前面分析ConcurrentHashMap的过程中可以发现,其要求key和value不能为空。实际上,不仅仅是ConcurrentHashMap,前面的Hash...

    冬天里的懒猫
  • ConcurrentHashMap源码学习

    既然有了HashMap为什么还会出现ConcurrentHashMap?同时ConcurrentHashMap具有什么优势?ConcurrentHashMap与...

    路行的亚洲
  • 猿实战16——承运商之搭建你的运费基石

    上几个章节,猿人君教会了你如何去实现前台类目的后台管理功能,今天我们一起来学习,如何实现承运商管理。

    山旮旯的胖子
  • android全局监控click事件的四种方式(小结)

    本文主要给大家分享如何在全局上去监听 click 点击事件,并做些通用处理或是拦截。使用场景可能就是具体的全局防快速重复点击,或是通用打点分析上报,用户行为监控...

    砸漏
  • 【Gym 100712B】Rock-Paper-Scissors

      对给定的对手的出拳顺序,如果只能按几个R,然后几个P,再几个S的顺序出拳(几个也可以是0个),那么求赢的方法有多少种。

    饶文津
  • Day3下午解题报告

    预计分数:20+40+30=90 实际分数:40+90+60=190 再次人品爆发&&手感爆发&&智商爆发 谁能告诉我为什么T1数据这么水。。 谁能告诉我为什么...

    attack
  • 编写优雅代码的最佳实践

    Robert Martin曾说过"在代码阅读中说脏话的频率是衡量代码质量额唯一标准"。同时,代码的写法应当使别人理解它所需的时间最小化,也就是说我们写的代码是给...

    木可大大
  • Java日志体系(commons-logging)Java日志系统学习

    贾博岩

扫码关注云+社区

领取腾讯云代金券