黑马程序员技术交流社区

标题: 【上海校区】多任务处理—Fork/Join框架（要点1） [打印本页]

---

作者: wuqiong    时间: 2018-6-12 11:05
标题: 【上海校区】多任务处理—Fork/Join框架（要点1）
1、工作过程概要

在开篇前，首先回答一个上篇文章中的一个问题。在上篇文章给出的第一个归并算法实例中提到，归并算法的算法效率和算法稳定性都是已知的排序算法中较好的，但是如果我们只是采用单线程的应用程序来运行它，那么算法性能还是不能全部发挥出来。文章还给了一张单线程运行归并排序算法时操作系统CPU的运行状态，如下图所示：

有的读者就询问，为什么单线程运行排序算法时，出现的效果是CPU1工作一段时间，然后CPU2又工作一段时间，接着其它CPU内核再分别工作一段时间的情况呢？不应该是某一个CPU内核连续工作直到运算过程完成吗？这是因为CPU计算资源完全由操作系统掌握，虽然应用程序中是一个线程在运行计算任务，且线程私有数据存储在CPU共享缓存中，而单线程任务工作时并不是一直连续占用CPU计算资源，而是会在操作系统协调下在挂起、就绪、运行状态下进行切换，而从上一次运行状态切换到下一次运行状态时，负责运行这个线程任务的CPU内核就可能会不同，所以就出现了类似上图所示的CPU切换效果。

Fork/Join Pool采用优良的设计、代码实现和硬件原子操作机制等多种思路保证其执行性能。其中包括（但不限于）：计算资源共享、高性能队列、避免伪共享、工作窃取机制等。本文（以及后续文章）试图和读者一起分析JDK1.8中Fork/Join Pool的源代码实现，去理解Fork/Join Pool是怎样工作的。当然这里要说明一下，起初本人在决定阅读Fork/Join归并计算相关类的源代码时（ForkJoinPool、WorkQueue、ForkJoinTask、RecursiveTask、ForkJoinWorkerThread等），并不觉得这部分代码比起LinkedList这样的类来说有多少难度， 但其中大量使用位运算和位运算技巧，有大量Unsafe原子操作。博主能力有限，确实不能在短时间内将所有代码一一详细解读，所以也希望各位读者能帮助笔者一同完善。

2、工作要点
2-1. Fork/Join Pool实例化

实际上在之前文章中给出的Fork/Join Pool使用实例中，我们使用的new ForkJoinPool()或者new ForkJoinPool(N)这些方式来进行操作，这并不是ForkJoinPool作者Doug Lea推荐的使用方式。在ForkJoinPool主类的注释说明中，有这样一句话：

A static commonPool() is available and appropriate for most applications. The common pool is used by any ForkJoinTask that is not explicitly submitted to a specified pool.
Using the common pool normally reduces resource usage (its threads are slowly reclaimed during periods of non-use, and reinstated upon subsequent use).
以上描述大致的中文解释是：ForkJoinPools类有一个静态方法commonPool()，这个静态方法所获得的ForkJoinPools实例是由整个应用进程共享的，并且它适合绝大多数的应用系统场景。使用commonPool通常可以帮助应用程序中多种需要进行归并计算的任务共享计算资源，从而使后者发挥最大作用（ForkJoinPools中的工作线程在闲置时会被缓慢回收，并在随后需要使用时被恢复），而这种获取ForkJoinPools实例的方式，才是Doug Lea推荐的使用方式。代码如下：

......
ForkJoinPool commonPool =  ForkJoinPool.commonPool();
......
1
2
3
通过阅读ForkJoinPool的代码我们可以发现ForkJoinPool中如何完成commonPool的初始化：

static {
  ......
  common = java.security.AccessController.doPrivileged
            (new java.security.PrivilegedAction<ForkJoinPool>() {
                public ForkJoinPool run() { return makeCommonPool(); }});
  // report 1 even if threads disabled
  int par = common.config & SMASK;
  commonParallelism = par > 0 ? par : 1;
  ......
}

......

// 这是主要的创建过程
private static ForkJoinPool makeCommonPool() {
   int parallelism = -1;
   ForkJoinWorkerThreadFactory factory = null;
   UncaughtExceptionHandler handler = null;
   // 可以通过在java程序启动时，指定这些参数的方式
   // 来完成并行等级，线程工厂，异常处理类的指定工作
   try {
       // 首先确认技术人员在启动应用程序时，是否指定了这些参数，来控制CommonPool的创建过程
       // ignore exceptions in accessing/parsing properties
       String pp =
       System.getProperty("java.util.concurrent.ForkJoinPool.common.parallelism");
       String fp =
       System.getProperty("java.util.concurrent.ForkJoinPool.common.threadFactory");
       String hp =
       System.getProperty("java.util.concurrent.ForkJoinPool.common.exceptionHandler");
       if (pp != null)
           parallelism = Integer.parseInt(pp);
       if (fp != null)
           factory = ((ForkJoinWorkerThreadFactory)ClassLoader.getSystemClassLoader().loadClass(fp).newInstance());
       if (hp != null)
           handler = ((UncaughtExceptionHandler)ClassLoader.getSystemClassLoader().loadClass(hp).newInstance());
   } catch (Exception ignore) {

   }

   // 没有在启动时指定以上参数也没关系，java会启动默认参数
   if (factory == null) {
       // 如果当前没有启动SecurityManager，安全策略管理器
       // 这时使用defaultForkJoinWorkerThreadFactory这个工厂对象
       // 它是java.util.concurrent.ForkJoinPool.DefaultForkJoinWorkerThreadFactory这个类的实例
       if (System.getSecurityManager() == null)
           factory = defaultForkJoinWorkerThreadFactory;
       else
           // use security-managed default
           factory = new InnocuousForkJoinWorkerThreadFactory();
   }

   // 如果并行等级小于0，并且当前应用程序可用CPU内核数为1
   // 那么设定parallelism并行等级为1
   if (parallelism < 0 && // default 1 less than #cores
       (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
       parallelism = 1;
   if (parallelism > MAX_CAP)
       parallelism = MAX_CAP;

   // 最后使用这个构造函数初始化commonPool
   return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE, "ForkJoinPool.commonPool-worker-");
}

以上代码片段中的中文注释是笔者加的，而英文注释是源代码自带的。对commonPool的初始化过程有Java security安全策略框架参与，doPrivileged方法为排除Java security安全策略框架的权限检查，而SecurityManager是Java security安全策略框架的管理器。一般情况下Java应用程序不会自动启动安全管理器，不过读者可以在Java应用程序启动时，使用-Djava.security.manager参数启动SecurityManager，或者在你的代码中通过System.setSecurityManager()方法显式设定一个。

当然，除了使用ForkJoinPool提供的commpool对象外，读者也可以直接通过ForkJoinPool提供的三种构造函数直接完成实例化，这三个可以同的构造分别是（以上构造函数的使用意义已经在之前的文章中讨论过了，这里就不再赘述了）：

public ForkJoinPool() {
    ......
}

public ForkJoinPool(int parallelism) {
    ......
}

public ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory,
                    UncaughtExceptionHandler handler, boolean asyncMode) {
    ......
}


2-2. 工作线程和工作队列

在本小节中我们主要讨论ForkJoinPool中处理ForkJoinTask任务及其子任务的情况，而ForkJoinPool处理Runnable或者Callable类型任务的情况将在后文讨论。ForkJoinPool中主要的工作线程，采用ForkJoinWorkerThread定义，其中有两个主要属性pool和workQueue：

public class ForkJoinWorkerThread extends Thread {
    ......
    // the pool this thread works in
    final ForkJoinPool pool;
    // work-stealing mechanics
    final ForkJoinPool.WorkQueue workQueue;
    ......
}

pool属性表示这个进行归并计算的线程所属的ForkJoinPool实例，workQueue属性是java.util.concurrent.ForkJoinPool.WorkQueue这个类的实例，它表示这个线程所使用的子任务待执行队列，而且可以被其它工作线程偷取任务。后者的内部是一个数组结构，并使用一些关键属性记录这个队列的实时状态，更具体的来说这个WorkQueue**是一个双端队列**。

Java中还有一组类似的双端队列顶层接口java.util.Deque、java.util.concurrent.BlockingDeque，但应该是出于实现细节的考虑，WorkQueue这个双端队列并没有实现这些接口。所谓双端队列，就是说队列中的元素（ForkJoinTask任务及其子任务）可以从一端入队出队，还可以从另一端入队出队。这个双端队列将用于支持ForkJoinPool的两种异步模型（asyncMode）：后进先出（LIFO_QUEUE）和先进先出（FIFO_QUEUE）。以下代码片段示例了WorkQueue类中定义的一些重要属性：

......

static final class WorkQueue {

    ......

    // 队列状态

    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)

    // 队列所属的ForkJoinPool（可能为空）

    // 注意，一个ForkJoinPool中会有多个执行线程，还会有比执行线程更多的（或一样多的）队列

    final ForkJoinPool pool;   // the containing pool (may be null)

    // 这个队列所属的归并计算工作线程。注意，工作队列也可能不属于任何工作线程

    final ForkJoinWorkerThread owner; // owning thread or null if shared

    // 记录当前正在进行join等待的其它任务

    volatile ForkJoinTask<?> currentJoin;  // task being joined in awaitJoin

    // 当前正在偷取的任务

    volatile ForkJoinTask<?> currentSteal; // mainly used by helpStealer

    ......

}

......

当ForkJoinWorkerThread需要向双端队列中放入一个新的待执行子任务时，会调用WorkQueue中的push方法。我们来看看这个方法的主要执行过程（请注意，源代码来自JDK1.8，它和JDK1.7中的实现有显著不同）：

/**

* Pushes a task. Call only by owner in unshared queues.  (The

* shared-queue version is embedded in method externalPush.)

*/

final void push(ForkJoinTask<?> task) {

    ForkJoinTask<?>[] a; ForkJoinPool p;

    int b = base, s = top, n;

    // 请注意，在执行task.fork时，触发push情况下，array不会为null

    // 因为在这之前workqueue中的array已经完成了初始化（在工作线程初始化时就完成了）

    if ((a = array) != null) {

    int m = a.length - 1;     // fenced write for task visibility

    // U常量是java底层的sun.misc.Unsafe操作类

    // 这个类提供硬件级别的原子操作

    // putOrderedObject方法在指定的对象a中，指定的内存偏移量的位置，赋予一个新的元素

    U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);

    // putOrderedInt方法对当前指定的对象中的指定字段，进行赋值操作

    // 这里的代码意义是将workQueue对象本身中的top标示的位置 + 1，

    U.putOrderedInt(this, QTOP, s + 1);

    if ((n = s - b) <= 1) {

        if ((p = pool) != null)

        // Tries to create or activate a worker if too few are active.

        // signalWork方法的意义在于，在当前活动的工作线程过少的情况下，创建新的工作线程

        p.signalWork(p.workQueues, this);

    }

    // 如果array的剩余空间不够了，则进行增加

    else if (n >= m)

        growArray();

    }

}

sun.misc.Unsafe操作类直接基于操作系统控制层在硬件层面上进行原子操作，它是ForkJoinPool高效性能的一大保证，类似的编程思路还体现在java.util.concurrent包中相当规模的类功能实现中。实际上sun.misc.Unsafe操作类在Java中有着举足轻重的地位，本专题的后续文章中会详细介绍sun.misc.Unsafe操作类，以及基于这个类实现的Java乐观锁机制。当ForkJoinWorkerThread需要从双端队列中取出下一个待执行子任务，就会根据设定的asyncMode调用双端队列的不同方法，代码概要如下所示：

// 试图从指定的队列中取出下一个待执行任务

final ForkJoinTask<?> nextTaskFor(WorkQueue w) {

    for (ForkJoinTask<?> t;;) {

        WorkQueue q; int b;

        // 该方法试图从“w”这个队列获取下一个待处理子任务

        if ((t = w.nextLocalTask()) != null)

            return t;

        // 如果没有获取到，则使用findNonEmptyStealQueue方法

        // 随机得到一个元素非空，并且可以进行任务窃取的存在于ForkJoinPool中的其它队列

        // 这个队列被记为“q”

        if ((q = findNonEmptyStealQueue()) == null)

            return null;

        // 试图从“q”这个队列base位处取出待执行任务

        if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null)

            return t;

    }

}

......

/**

* Takes next task, if one exists, in order specified by mode.

*/

final ForkJoinTask<?> nextLocalTask() {

    // 如果asyncMode设定为后进先出（LIFO）

    // 则使用pop()从双端队列的前端取出任务

    // 否则就是先进先出模式（LIFO），使用poll()从双端队列的后端取出任务

    return (config & FIFO_QUEUE) == 0 ? pop() : poll();

}

......

===========接下文

---

欢迎光临 黑马程序员技术交流社区 (http://bbs.itheima.com/)

黑马程序员IT技术论坛 X3.2