专栏首页做不甩锅的后端java线程池(六):ForkJoinPool源码分析之二(WorkQueue源码)

java线程池(六):ForkJoinPool源码分析之二(WorkQueue源码)

在前面介绍了ForkJoinPool的骨架源码之后,我们来看看ForkJoinPool的核心组成。WorkQueue的源码。

1.类结构及其成员变量

1.1 类结构和注释

WorkQueue是ForkJoinPool的核心内部类,是一个Contented修饰的静态内部类。

/**
 * Queues supporting work-stealing as well as external task
 * submission. See above for descriptions and algorithms.
 * Performance on most platforms is very sensitive to placement of
 * instances of both WorkQueues and their arrays -- we absolutely
 * do not want multiple WorkQueue instances or multiple queue
 * arrays sharing cache lines. The @Contended annotation alerts
 * JVMs to try to keep instances apart.
 */
@sun.misc.Contended
static final class WorkQueue {
}

其注释大意为: workQUeue是一个支持任务窃取和外部提交任务的队列,其实现参考ForkJoinPool描述的算法。在大多数平台上的性能对工作队列及其数组的实例都非常敏感。我们不希望多个工作队列的实例和多个队列数组共享缓存。@Contented注释用来提醒jvm将workQueue在执行的时候与其他对象进行区别。 在前面学过@Contented的内容,实际上就是采用内存对齐的方式,保证WorkQueue在执行的时候,其前后不会有其他对象干扰。

1.2 常量

在WorkQueue中有两个重要的常量,分别是INITIAL_QUEUE_CAPACITY和MAXIMUM_QUEUE_CAPACITY。

1.2.1 MAXIMUM_QUEUE_CAPACITY

MAXIMUM_QUEUE_CAPACITY注释如下:

/**
 * Capacity of work-stealing queue array upon initialization.
 * Must be a power of two; at least 4, but should be larger to
 * reduce or eliminate cacheline sharing among queues.
 * Currently, it is much larger, as a partial workaround for
 * the fact that JVMs often place arrays in locations that
 * share GC bookkeeping (especially cardmarks) such that
 * per-write accesses encounter serious memory contention.
 */
static final int INITIAL_QUEUE_CAPACITY = 1 << 13;

这是工作窃取队列的初始化容量,这个容量必须是2的幂,而且至少是4。理论上应该比4更大,以减少在CPU执行的时候多个队列进行共享内存的情况。这个值目前是远大于4的。做为一种局部解决方案,jvm经常将数组放在共享GC的记录中,尤其是cardmarks,这样在每次访问的过程中都会出现严重的内存争用。

1.2.2 MAXIMUM_QUEUE_CAPACITY

MAXIMUM_QUEUE_CAPACITY注释如下:

/**
 * Maximum size for queue arrays. Must be a power of two less
 * than or equal to 1 << (31 - width of array entry) to ensure
 * lack of wraparound of index calculations, but defined to a
 * value a bit less than this to help users trap runaway
 * programs before saturating systems.
 */
static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M

MAXIMUM_QUEUE_CAPACITY是队列支持的最大容量,必须是2的幂小于或等于1<<(31-数组项的宽度),但定义为一个略小于此值的值,以帮助用户在饱和系统之前捕获失控的程序。

1.3 成员变量

成员变量区如下:

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

这些成员变量整理后见下表:

变量名

类型

说明

scanState

volatile int

记录队列的扫描信息,当小于0的时候,不活跃,当为奇数的时候,则处于扫描状态中

stackPred

int

前一个ctl

nsteals

int

窃取的次数

hint

用于窃取线程进行随机选择被窃取的初始化索引的计算值

config

int

用于存储线程池的index和model

qlock

volatile int

1 表示锁定,小于0的时候表示结束,线程池的其他状态下这个值都应该为0

base

volatile int

下一个进行poll操作的索引

top

int

下一个pull操作的索引

array

ForkJoinTask<?>[]

存放task的数组,初始化的时候不会进行分配,采用懒加载的方式

pool

final ForkJoinPool

指向ForkJoinPool的指针

owner

final ForkJoinWorkerThread

工作队列的所有者线程,如果为共享任务队列则没有所有者,这个值为空

parker

volatile Thread

owoner线程在调用过程中如果出现park阻塞,则这个变量指向owoner,反之为空

currentJoin

volatile ForkJoinTask<?>

正在等待Join的任务

currentSteal

volatile ForkJoinTask<?>

主要用于帮助窃取

2.构造函数

WorkQueue就一个构造函数:

 WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner) {
        this.pool = pool;
        this.owner = owner;
        // Place indices in the center of array (that is not yet allocated)
        base = top = INITIAL_QUEUE_CAPACITY >>> 1;
}

在这个构造函数中,只会指定pool和owoner,如果该队列是共享队列,那么owoner此时是空的。此外,base和top两个指针分别都指向了数组的中值,这个值是初始化容量右移一位。 那么结合前面的代码,实际上初始化的时候,数组的长度为8192,那么base=top=4096。 这个数组在构造函数被调用之后初始化如下:

3.重要的方法

3.1 push

/**
 * Pushes a task. Call only by owner in unshared queues.  (The
 * shared-queue version is embedded in method externalPush.)
 *
 * @param task the task. Caller must ensure non-null.
 * @throws RejectedExecutionException if array cannot be resized
 */
final void push(ForkJoinTask<?> task) {
    ForkJoinTask<?>[] a; ForkJoinPool p;
    int b = base, s = top, n;
    //判断array不为空
    if ((a = array) != null) {    // ignore if queue removed
        //m为最高为位置的index
        int m = a.length - 1;     // fenced write for task visibility
        //将task采用cas的方式,put到数组中的top+1的位置,下面代码的一大堆操作实际上是与cas相关的,需要计算类对象中的偏移量,如果我们不用usafe类,那么这个地方就会非常简单。
        U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);
        //采用cas的方式,将top的指针加1
        U.putOrderedInt(this, QTOP, s + 1);
        //如果n小于等于1则 且poll不为空 则触发worker窃取或者产生新的worker
        if ((n = s - b) <= 1) {
            if ((p = pool) != null)
                p.signalWork(p.workQueues, this);
        }
        //如果n大于等于了m 则说明需要扩容了
        else if (n >= m)
            growArray();
    }
}

这个push方法是提供给工作队列自己push任务来使用的,共享队列push任务是在外部externalPush和externalSubmit等方法来进行初始化和push。 这里需要注意的是,当队列中的任务数小于1的时候,才会调用signalWork,这个地方一开始并不理解,实际上,我们需要注意的是,这个方法是专门提供给工作队列来使用的,那么这个条件满足的时候,说明工作队列空闲。如果这个条件不满足,那么工作队列中有很多任务需要工作队列来处理,就不会触发对这个队列的窃取操作。

3.2 growArray

这是扩容的方法。实际上这个方法有两个作用,首先是初始化,其次是判断,是否需要扩容,如果需要扩容则容量加倍。

/**
 * Initializes or doubles the capacity of array. Call either
 * by owner or with lock held -- it is OK for base, but not
 * top, to move while resizings are in progress.
 */
final ForkJoinTask<?>[] growArray() {
    //旧的数组 oldA
    ForkJoinTask<?>[] oldA = array;
    //如果oldA不为空,则size就为oldA的长度*2,反之说明数组没有被初始化,那么长度就应该为初始化的长度8192
    int size = oldA != null ? oldA.length << 1 : INITIAL_QUEUE_CAPACITY;
    //如果size比允许的最大容量还大,那么此时会抛出异常
    if (size > MAXIMUM_QUEUE_CAPACITY)
        throw new RejectedExecutionException("Queue capacity exceeded");
    int oldMask, t, b;
    //array a 为根据size new出来的一个新的数组
    ForkJoinTask<?>[] a = array = new ForkJoinTask<?>[size];
    //如果oldA不为空且其长度大于等于0为有效数组,且top-base大于0 说明不为空
    if (oldA != null && (oldMask = oldA.length - 1) >= 0 &&
        (t = top) - (b = base) > 0) {
        //按size定义掩码
        int mask = size - 1;
        //从旧的数组中poll全部task,然后push到新的array中
        do { // emulate poll from old array, push to new array
            ForkJoinTask<?> x;
            //采用unsafe操作
            int oldj = ((b & oldMask) << ASHIFT) + ABASE;
            int j    = ((b &    mask) << ASHIFT) + ABASE;
            //实际上直接进行的内存对象copy,这样效率比循环调用push和poll要高很多
            x = (ForkJoinTask<?>)U.getObjectVolatile(oldA, oldj);
            //判断  x不为空 则使用unsafe进行操作
            if (x != null &&
                U.compareAndSwapObject(oldA, oldj, x, null))
                U.putObjectVolatile(a, j, x);
        } while (++b != t);
    }
    //返回新的数组
    return a;
}

需要注意的是,这个方法一旦调用进行扩容之后,无论是来自于外部push操作触发,还是有工作线程worker触发,都将被锁定,之后,不能移动top指针,但是base指针是可以移动的。这也就是说,一旦处于扩容的过程中,就不能新增task,但是可以从base进行消费,这就只支持FIFO。因此同步模式将在此时被阻塞。

3.3 pop

同样,pop操作也仅限于工作线程,对于共享对立中则不允许使用pop方法。这个方法将按LIFO后进先出的方式从队列中。

/**
 * Takes next task, if one exists, in LIFO order.  Call only
 * by owner in unshared queues.
 */
final ForkJoinTask<?> pop() {
    ForkJoinTask<?>[] a; ForkJoinTask<?> t; int m;
    //如果array不为空切长度大于0
    if ((a = array) != null && (m = a.length - 1) >= 0) {
        //循环,s为top的指针减1,即top减1之后要大于0 也就是说要存在task
        for (int s; (s = top - 1) - base >= 0;) {
            //计算unsafe的偏移量 得到s的位置
            long j = ((m & s) << ASHIFT) + ABASE;
            //如果这个索引处的对象为空,则退出
            if ((t = (ForkJoinTask<?>)U.getObject(a, j)) == null)
                break;
            //反之用usafe的方法将这个值取走,之后返回,并更新top的指针
            if (U.compareAndSwapObject(a, j, t, null)) {
                U.putOrderedInt(this, QTOP, s);
                return t;
            }
        }
    }
    return null;
}

pop方法,这是仅限于owoner调用的方法,将从top指针处取出task。这个方法对于整个队列是LIFO的方式。

3.4 poll

poll方法将从队列中按FIFO的方式取出task。

/**
 * Takes next task, if one exists, in FIFO order.
 */
final ForkJoinTask<?> poll() {
    ForkJoinTask<?>[] a; int b; ForkJoinTask<?> t;
    //判断 base-top小于0说明存在task 切array不为空
    while ((b = base) - top < 0 && (a = array) != null) {
        //计算出unsafe操作的索引 实际上就是拿到b
        int j = (((a.length - 1) & b) << ASHIFT) + ABASE;
        //之后拿到这个task 用volatile的方式
        t = (ForkJoinTask<?>)U.getObjectVolatile(a, j);
        //之后如果base和b相等
        if (base == b) {
           //如果拿到的task不为空
            if (t != null) {
                //那么将这个位置的元素移除 base+1 然后返回t
                if (U.compareAndSwapObject(a, j, t, null)) {
                    base = b + 1;
                    return t;
                }
            }
            //在上述操作之后,如果base比top小1说明已经为空了 直接退出循环
            else if (b + 1 == top) // now empty
                break;
        }
    }
    //默认返回null
    return null;
}

3.5 pollAt

这个方法将采用FIFO的方式,从 队列中获得task。

/**
 * Takes a task in FIFO order if b is base of queue and a task
 * can be claimed without contention. Specialized versions
 * appear in ForkJoinPool methods scan and helpStealer.
 */
final ForkJoinTask<?> pollAt(int b) {
    ForkJoinTask<?> t; ForkJoinTask<?>[] a;
    //数组不为空
    if ((a = array) != null) {
        //计算索引b的位置
        int j = (((a.length - 1) & b) << ASHIFT) + ABASE;
        //如果此处的task不为空,则将此处置为null然后将对象task返回
        if ((t = (ForkJoinTask<?>)U.getObjectVolatile(a, j)) != null &&
            base == b && U.compareAndSwapObject(a, j, t, null)) {
            base = b + 1;
            return t;
        }
    }
    return null;
}

通常情况下,b指的是队列的base指针。那么从底部获取元素就能实现FIFO。特殊的版本出现在scan和helpStealer中用于对工作队列的窃取操作的实现。

3.6 nextLocalTask

/**
 * Takes next task, if one exists, in order specified by mode.
 */
final ForkJoinTask<?> nextLocalTask() {
    return (config & FIFO_QUEUE) == 0 ? pop() : poll();
}

这个方法中对之前的MODE会起作用,如果是FIFO则用pop方法,反之则用poll方法获得下一个task。

3.7 peek

/**
 * Returns next task, if one exists, in order specified by mode.
 */
final ForkJoinTask<?> peek() {
    ForkJoinTask<?>[] a = array; int m;
    //判断数组的合法性
    if (a == null || (m = a.length - 1) < 0)
        return null;
    //根据mode决定从top还是base处获得task
    int i = (config & FIFO_QUEUE) == 0 ? top - 1 : base;
    int j = ((i & m) << ASHIFT) + ABASE;
    //返回获得的task
    return (ForkJoinTask<?>)U.getObjectVolatile(a, j);
}

peek则根据之前的mode定义,从队列的前面或者后面取得task。

3.8 tryUnpush

/**
 * Pops the given task only if it is at the current top.
 * (A shared version is available only via FJP.tryExternalUnpush)
*/
final boolean tryUnpush(ForkJoinTask<?> t) {
    ForkJoinTask<?>[] a; int s;
    //判断数组的合法性
    if ((a = array) != null && (s = top) != base &&
        //将top位置的task与t比较,如果相等则将其改为null
        U.compareAndSwapObject
        (a, (((a.length - 1) & --s) << ASHIFT) + ABASE, t, null)) {
        //将top减1
        U.putOrderedInt(this, QTOP, s);
        //返回操作成功
        return true;
    }
    //默认返回失败
    return false;
}

这个方法是将之前push的任务撤回。这个操作仅仅只有task位于top的时候操能成功。

3.9 runTask

在之前的文章分析外部提交task的时候,就提到了这个方法。实际上是runWorker调用的。 也就是说,线程在启动之后,一旦worker获取到task,就会运行。

/**
 * Executes the given task and any remaining local tasks.
 */
final void runTask(ForkJoinTask<?> task) {
    //task不为空
    if (task != null) {
        //扫描状态标记为busy 那么说明当前的worker正在处理本地任务   此时这个操作会将scanState改为0
        scanState &= ~SCANNING; // mark as busy
        //执行这个task
        (currentSteal = task).doExec();
        //释放已执行任务的内存
        U.putOrderedObject(this, QCURRENTSTEAL, null); // release for GC
        //执行其他本地的task
        execLocalTasks();
        ForkJoinWorkerThread thread = owner;
        //增加增加steals的次数
        if (++nsteals < 0)      // collect on overflow
            transferStealCount(pool);
        //将scanState改为1 这样就变得活跃可以被其他worker scan
        scanState |= SCANNING;
        //如果thread不为null说明为worker线程 则调用后续的exec方法
        if (thread != null)
            thread.afterTopLevelExec();
    }
}

3.10 execLocalTasks

调用这个方法,运行队列中的全部task,如果采用了LIFO模式,则调用pollAndExecAll,这是另外一种实现方法。直到将队列都执行到empty

/**
 * Removes and executes all local tasks. If LIFO, invokes
 * pollAndExecAll. Otherwise implements a specialized pop loop
 * to exec until empty.
 */
final void execLocalTasks() {
    int b = base, m, s;
    //拿到数组
    ForkJoinTask<?>[] a = array;
    //如果b-s小于0说明存在task,a不为空,切a的长度大于0 这均是检测方法的合法性
    if (b - (s = top - 1) <= 0 && a != null &&
        (m = a.length - 1) >= 0) {
        //如果没有采用FIFO的mode  那么一定是LIFO 则从top处开始
        if ((config & FIFO_QUEUE) == 0) {
           //开始循环
            for (ForkJoinTask<?> t;;) {
               //从top开始取出task
                if ((t = (ForkJoinTask<?>)U.getAndSetObject
                     (a, ((m & s) << ASHIFT) + ABASE, null)) == null)
                    break;
                //修改top
                U.putOrderedInt(this, QTOP, s);
                //执行task
                t.doExec();
                //如果没有任务的了 则退出
                if (base - (s = top - 1) > 0)
                    break;
            }
        }
        else
           //FIFO的方式调用pollAndExecAll
            pollAndExecAll();
    }
}

3.11 pollAndExecAll

此方法将用poll,FIFO的方式获得task并执行。

  final void pollAndExecAll() {
        for (ForkJoinTask<?> t; (t = poll()) != null;)
            t.doExec();
    }

可见,当通过workQueue中调用runTask的方法的时候,会将这个队列的scanState状态修改为0,之后将这个队列中的全部task根据定义的mode全部消费完毕。

3.12 tryRemoveAndExec

从注释中可知,这个方法仅仅供awaitJoin方法调用,在await的过程中,将task从workQueue中移除并执行。

/**
 * If present, removes from queue and executes the given task,
 * or any other cancelled task. Used only by awaitJoin.
 *
 * @return true if queue empty and task not known to be done
 */
final boolean tryRemoveAndExec(ForkJoinTask<?> task) {
    ForkJoinTask<?>[] a; int m, s, b, n;
    //判断数组的合法性 task不能为空
    if ((a = array) != null && (m = a.length - 1) >= 0 &&
        task != null) {
        //循环  n为task的数量,必须大于0
        while ((n = (s = top) - (b = base)) > 0) {
            //死循环 从top遍历到base
            for (ForkJoinTask<?> t;;) {      // traverse from s to b
                long j = ((--s & m) << ASHIFT) + ABASE;
                if ((t = (ForkJoinTask<?>)U.getObject(a, j)) == null)
                    return s + 1 == top;     // shorter than expected
                //如果task处于top位置
                else if (t == task) {
                    boolean removed = false;
                    if (s + 1 == top) {      // pop
                        //pop的方式获取task  然后替换为null
                        if (U.compareAndSwapObject(a, j, task, null)) {
                            U.putOrderedInt(this, QTOP, s);
                            removed = true;
                        }
                    }
                    //用emptytask代替
                    else if (base == b)      // replace with proxy
                        removed = U.compareAndSwapObject(
                            a, j, task, new EmptyTask());
                    //如果remove成功 则执行这个task
                    if (removed)
                        task.doExec();
                    break;
                }
                //如果task的status为负数 切 top=s=1
                else if (t.status < 0 && s + 1 == top) {
                    //移除
                    if (U.compareAndSwapObject(a, j, t, null))
                        U.putOrderedInt(this, QTOP, s);
                    break;                  // was cancelled
                }
                if (--n == 0)
                    return false;
            }
            if (task.status < 0)
                return false;
        }
    }
    return true;
}

3.13 popCC

如果pop CountedCompleter。这方法支持共享和worker的队列,但是仅仅通过helpComplete调用。 CountedCompleter是jdk1.8中新增的一个ForkJoinTask的一个实现类。

/**
 * Pops task if in the same CC computation as the given task,
 * in either shared or owned mode. Used only by helpComplete.
 */
final CountedCompleter<?> popCC(CountedCompleter<?> task, int mode) {
    int s; ForkJoinTask<?>[] a; Object o;
    //判断队列数组合法性
    if (base - (s = top) < 0 && (a = array) != null) {
        //从top处开始
        long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE;
        //如果获的的task不为null
        if ((o = U.getObjectVolatile(a, j)) != null &&
            //且为CountedCompleter对象
            (o instanceof CountedCompleter)) {
            //转换为CountedCompleter
            CountedCompleter<?> t = (CountedCompleter<?>)o;
            //死循环
            for (CountedCompleter<?> r = t;;) {
                //如果task与获得的r相等为同一对象
                if (r == task) {
                    //如果mode小于0 
                    if (mode < 0) { // must lock
                         //cas的方式加锁
                        if (U.compareAndSwapInt(this, QLOCK, 0, 1)) {
                            //将这个对象清除 并修改top后解锁
                            if (top == s && array == a &&
                                U.compareAndSwapObject(a, j, t, null)) {
                                U.putOrderedInt(this, QTOP, s - 1);
                                U.putOrderedInt(this, QLOCK, 0);
                                //返回t
                                return t;
                            }
                            //解锁
                            U.compareAndSwapInt(this, QLOCK, 1, 0);
                        }
                    }
                    else if (U.compareAndSwapObject(a, j, t, null)) {
                        U.putOrderedInt(this, QTOP, s - 1);
                        return t;
                    }
                    break;
                }
                else if ((r = r.completer) == null) // try parent
                    break;
            }
        }
    }
    return null;
}

3.14 pollAndExecCC

pollAndExecCC 。窃取并运行与给定任务相同CountedCompleter计算任务(如果存在),并且可以在不发生争用的情况下执行该任务。否则,返回一个校验和/控制值,供helpComplete方法使用。

/**
 * Steals and runs a task in the same CC computation as the
 * given task if one exists and can be taken without
 * contention. Otherwise returns a checksum/control value for
 * use by method helpComplete.
 *
 * @return 1 if successful, 2 if retryable (lost to another
 * stealer), -1 if non-empty but no matching task found, else
 * the base index, forced negative.
 */
final int pollAndExecCC(CountedCompleter<?> task) {
    int b, h; ForkJoinTask<?>[] a; Object o;
    //判断array的合法性
    if ((b = base) - top >= 0 || (a = array) == null)
        h = b | Integer.MIN_VALUE;  // to sense movement on re-poll
    else {
        //从base开始获得task
        long j = (((a.length - 1) & b) << ASHIFT) + ABASE;
        if ((o = U.getObjectVolatile(a, j)) == null)
            h = 2;                  // retryable
        else if (!(o instanceof CountedCompleter))
            h = -1;                 // unmatchable
        else {
            CountedCompleter<?> t = (CountedCompleter<?>)o;
            //死循环
            for (CountedCompleter<?> r = t;;) {
                if (r == task) {
                    if (base == b &&
                        U.compareAndSwapObject(a, j, t, null)) {
                        base = b + 1;
                        t.doExec();
                        h = 1;      // success
                    }
                    else
                        h = 2;      // lost CAS
                    break;
                }
                else if ((r = r.completer) == null) {
                    h = -1;         // unmatched
                    break;
                }
            }
        }
    }
    return h;
}

4.总结

本文对workQueue的源码进行了分析,我们需要注意的是,对于workQueue,定义了三个操作,分别是push,poll和pop。

  • push

主要是操作top指针,将top进行移动。

  • poll 如果top和base不等,则说明队列有值,可以消费,那么poll就从base指针处开始消费。这个方法实现了队列的FIFO。

消费之后对base进行移动。

  • pop 同样,还可以从top开始消费,这就是pop。这个方法实际上实现了对队列的LIFO。

消费之后将top减1。

以上就是这三个方法对应的操作。但是我们还需要注意的是,在所有的unsafe操作中,通过cas进行设置或者获得task的时候,还有一个掩码。这个非常重要。 我们可以看在push方法中:

 int m = a.length - 1;
 U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);

在扩容的方法growArray中我们可以知道。每次扩容都是采用左移的方式来进行,这样就保证了数组的长度为2的幂。 在这里,m=a.length-1,那就说明,m实际上其二进制格式将会有效位都为1,这个数字就可以做为掩码。当m再与s取&计算的时候。可以想象,s大于m的部分将被去除,只会保留比m小的部分。那么实际上,这就等价于,当我们一直再push元素到数组中的时候,实际上就从数组的索引底部开始:

参考上面这个过程,也就是说,实际上这个数组,base和top实际指向的index并不重要。只有二者的相对位移才是重要的。这有点类似与RingBuffer的数据结构,但是还是有所不同。也就是说这个数组实际上是不会被浪费的。之前有很多不理解的地方,为什么top减去base可能出现负数。那么这样实际上就会导致负数的产生。 这样的话,如果我们采用异步模式,asyncMode为true的时候,workQueue则会采用FIFO_QUEUE的model,这样workQueue本身就使用的时poll方法。反之如果使用LIFO_QUEUE的同步模式,则workQueue使用pop方法。默认情况下采用同步模式。同步的时候workQueue的指针都围绕在数组的初始化的中间位置波动。而共享队列则会一直循环。 至此,我们分析了workQueue的源码,对其内部实现的双端队列本身的操作进行了分析。为什么作者会自己实现一个Deque,而不是使用juc中已存在的容器。这就是因为这个队列全程都是采用Unsafe来实现的,在开篇作者也说了,需要@Contented修饰,就是为了避免缓存的伪代共享。这样来实现一个高效的Deque,以供ForkJoinPool来操作。 这与学习ConcurrentHashMap等容器的源码一样,可以看出作者为了性能的优化,采用了很多独特的方式来实现。这些地方都是我们值得学习和借鉴之处。这也是ForkJoin性能高效的关键。在作者的论文中也可以看出,java的实现,由于抽象在jvm之上,性能比c/c++的实现要低很多。这也是作者尽可能将性能做到最优的原因之一。

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

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 聊聊java中的哪些Map:(九)TreeMap源码分析

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

    冬天里的懒猫
  • 聊聊java中的哪些Map:(二)HashMap中的TreeNode

    TreeNode则是hashMap树化之后,组成树的基本节点。需要注意的是,TreeNode继承了LiknedHashMap.Entry ,LinkedHas...

    冬天里的懒猫
  • 09 Confluent_Kafka权威指南 第九章:管理kafka集群

    Kafka提供了几个命令行接口实用程序,他们对于kafka集群的配置管理非常有用。这些工具是通过java来实现的,并提供了一组脚本来调用这些类。这些工具提供了基...

    冬天里的懒猫
  • 代码写成这样,老板把我开除了,最让人感到无可奈何的代码,你可千万别这么写!

    后面有人挖出来这段神逻辑,大家真是哭晕在厕所了,还好,抢购买手机这种我从来不参加。。

    乔戈里
  • java判断字符串为空

    if(s == null ||"".equals(s)); 方法二: 比较字符串长度,...

    斯文的程序
  • python3.5程序设计

    Python语言的特点有: 易于学习:语法和结构简单,学习起来更加简单。 易于阅读:Python代码定义更清晰。 易于维护:源代码是相当容易维护的。 广泛的标准...

    py3study
  • dubbo源码——服务提供者的服务暴露过程(一)

    <dubbo:xxx/>标签是Spring的自定义标签,可以查看dubbo.jar包下的META-INF->spring.handlers

    冰枫
  • 由null出发-看看为何mysql这么排斥null

    在表结构设计时我们经常会面对是否要加空值约束、默认值处理等问题,当向前人经验中检索时,经常会看到不建议为null,强刷面经时not null也经常被归结到优化建...

    邹志全
  • 使用Optional来减少null检查

    平常我们使用null检查在项目中简直太常见了,从数据库中查询到的数据可能不存在返回null,service中处理中发现不存在返回一个null,在互相调用的时候每...

    Dylan Liu
  • Java中有关Null的9件事

    对于Java程序员来说,null是令人头痛的东西。时常会受到空指针异常(NPE)的骚扰。连Java的发明者都承认这是他的一项巨大失误。Java为 什么要保留nu...

    哲洛不闹

扫码关注云+社区

领取腾讯云代金券