Fork/Join 框架

Fork/Join 框架是 Java7 提供了的一个用于并行执行任务的框架， 是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。
Fork就是把一个大任务切分为若干子任务并行的执行，Join 就是合并这些子任务的执行结果，最后得到这个大任务的结果。比如计算1+2+…..＋10000，可以分割成 10 个子任务，每个子任务分别对 1000 个数进行求和，最终汇总这 10 个子任务的结果。如下图所示：

Fork/Jion特性：

1. ForkJoinPool 不是为了替代 ExecutorService，而是它的补充，在某些应用场景下性能比ExecutorService 更好。（见 Java Tip: When to use ForkJoinPool vs ExecutorService ）
   2. ForkJoinPool 主要用于实现“分而治之”的算法，特别是分治之后递归调用的函数，例如 quick sort 等。
   3. ForkJoinPool 最适合的是计算密集型的任务，如果存在 I/O，线程间同步，sleep() 等会造成线程长时间阻塞的情况时，最好配合使用 ManagedBlocker。

   工作窃取算法

   工作窃取（work-stealing）算法是指某个线程从其他队列里窃取任务来执行。我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。
   工作窃取算法的优点是充分利用线程进行并行计算，并减少了线程间的竞争，其缺点是在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且消耗了更多的系统资源，比如创建多个线程和多个双端队列。
   
   1. ForkJoinPool 的每个工作线程都维护着一个工作队列（WorkQueue），这是一个双端队列（Deque），里面存放的对象是任务（ForkJoinTask）。
   2. 每个工作线程在运行中产生新的任务（通常是因为调用了 fork()）时，会放入工作队列的队尾，并且工作线程在处理自己的工作队列时，使用的是 LIFO 方式，也就是说每次从队尾取出任务来执行。
   3. 每个工作线程在处理自己的工作队列同时，会尝试窃取一个任务（或是来自于刚刚提交到 pool 的任务，或是来自于其他工作线程的工作队列），窃取的任务位于其他线程的工作队列的队首，也就是说工作线程在窃取其他工作线程的任务时，使用的是FIFO 方式。
   4. 在遇到 join() 时，如果需要 join 的任务尚未完成，则会先处理其他任务，并等待其完成。
   5. 在既没有自己的任务，也没有可以窃取的任务时，进入休眠。

   fork/join的使用

   ForkJoinTask：我们要使用 ForkJoin 框架，必须首先创建一个 ForkJoin 任务。它提供在任务中执行fork() 和 join() 操作的机制，通常情况下我们不需要直接继承ForkJoinTask类，而只需要继承它的子类，Fork/Join 框架提供了以下两个子类：

• RecursiveAction：用于没有返回结果的任务。(如写数据到磁盘，然后就退出了。一个RecursiveAction可以把自己的工作分成更小的几块， 这样就可以由独立的线程或CPU执行。可以通过继承来实现一个RecursiveAction)
• RecursiveTask ：用于有返回结果的任务。(可以将自己的工作分割为若干更小任务，并将这些子任务的执行合并到一个集体结果。 可以有几个水平的分割和合并)

CountedCompleter： 在任务完成执行后会触发执行一个自定义的钩子函数

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

fork/join框架原理

异常处理

ForkJoinTask 在执行的时候可能会抛出异常，但是我们没办法在主线程里直接捕获异常，所以 ForkJoinTask 提供了 isCompletedAbnormally() 方法来检查任务是否已经抛出异常或已经被取消了，并且可以通过 ForkJoinTask 的 getException 方法获取异常。示例如下

if(task.isCompletedAbnormally()){
    System.out.println(task.getException());
}

etException 方法返回 Throwable 对象，如果任务被取消了则返回CancellationException。如果任务没有完成或者没有抛出异常则返回 null。

ForkJoinPool构造函数

其完整构造方法如下

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：并行度(the parallelism level)，默认情况下跟机器的cpu个数保持一致，使用Runtime.getRuntime().availableProcessors()可以得到机器运行时可用的CPU个数。
②factory：创建新线程的工厂（ the factory for creating new threads）。默认情况下使用ForkJoinWorkerThreadFactory defaultForkJoinWorkerThreadFactory。
③handler：线程异常情况下的处理器（Thread.UncaughtExceptionHandler handler），该处理器在线程执行任务时由于某些无法预料到的错误而导致任务线程中断时进行一些处理，默认情况为null。
④asyncMode：这个参数要注意，在ForkJoinPool中，每一个工作线程都有一个独立的任务队列，asyncMode表示工作线程内的任务队列是采用何种方式进行调度，可以是先进先出FIFO，也可以是后进先出LIFO。如果为true，则线程池中的工作线程则使用先进先出方式进行任务调度，默认情况下是false。

ForkJoinTask fork 方法

fork() 做的工作只有一件事，既是把任务推入当前工作线程的工作队列里。可以参看以下的源代码：

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;
}

ForkJoinTask join 方法

join() 的工作则复杂得多，也是 join() 可以使得线程免于被阻塞的原因——不像同名的 Thread.join()。
1. 检查调用 join() 的线程是否是 ForkJoinThread 线程。如果不是（例如 main 线程），则阻塞当前线程，等待任务完成。如果是，则不阻塞。
2. 查看任务的完成状态，如果已经完成，直接返回结果。
3. 如果任务尚未完成，但处于自己的工作队列内，则完成它。
4. 如果任务已经被其他的工作线程偷走，则窃取这个小偷的工作队列内的任务（以 FIFO 方式），执行，以期帮助它早日完成欲 join 的任务。
5. 如果偷走任务的小偷也已经把自己的任务全部做完，正在等待需要 join 的任务时，则找到小偷的小偷，帮助它完成它的任务。
6. 递归地执行第5步。将上述流程画成序列图的话就是这个样子：

ForkJoinPool.submit 方法

public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
    if (task == null)
        throw new NullPointerException();
    //提交到工作队列
    externalPush(task);
    return task;
}

ForkJoinPool 自身拥有工作队列，这些工作队列的作用是用来接收由外部线程（非 ForkJoinThread 线程）提交过来的任务，而这些工作队列被称为 submitting queue 。submit() 和 fork() 其实没有本质区别，只是提交对象变成了 submitting queue而已（还有一些同步，初始化的操作）。submitting queue 和其他 work queue 一样，是工作线程”窃取“的对象，因此当其中的任务被一个工作线程成功窃取时，就意味着提交的任务真正开始进入执行阶段。

Fork/Join框架执行流程

ScheduledThreadPoolExecutor

定时线程池类的类结构图

它用来处理延时任务或定时任务

它接收SchduledFutureTask类型的任务，是线程池调度任务的最小单位，有三种提交任务的方式：

1. schedule
2. scheduledAtFixedRate
3. scheduledWithFixedDelay

它采用DelayQueue存储等待的任务

1. DelayQueue内部封装了一个PriorityQueue，它会根据time的先后时间排序，若time相同则根据sequenceNumber排序；
2. DelayQueue也是一个无界队列；

   SchduledFutureTask

   SchduledFutureTask接收的参数(成员变量)：
   1. private long time：任务开始的时间
   2. private final long sequenceNumber;：任务的序号
   3. private final long period：任务执行的时间间隔
   工作线程的执行过程：

• 工作线程会从DelayQueue取已经到期的任务去执行；
• 执行结束后重新设置任务的到期时间，再次放回DelayQueue

ScheduledThreadPoolExecutor会把待执行的任务放到工作队列DelayQueue中，DelayQueue封装了一个PriorityQueue，PriorityQueue会对队列中的ScheduledFutureTask进行排序，具体的排序算法实现如下：

public int compareTo(Delayed other) {
    if (other == this) // compare zero if same object
        return 0;
    if (other instanceof ScheduledFutureTask) {
        ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
        long diff = time - x.time;
        if (diff < 0)
            return -1;
        else if (diff > 0)
            return 1;
        else if (sequenceNumber < x.sequenceNumber)
            return -1;
        else
            return 1;
    }
    long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
    return (diff < 0) ? ­1 : (diff > 0) ? 1 : 0;
}

1. 首先按照time排序，time小的排在前面，time大的排在后面；
   2. 如果time相同，按照sequenceNumber排序，sequenceNumber小的排在前面，sequenceNumber大的排在后面，换句话说，如果两个task的执行时间相同，优先执行先提交的task。

   SchduledFutureTask之run方法实现

   run方法是调度task的核心，task的执行实际上是run方法的执行。

   public void run() {
    boolean periodic = isPeriodic();
    //如果当前线程池已经不支持执行任务，则取消
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    //如果不需要周期性执行，则直接执行run方法然后结束
    else if (!periodic)
        ScheduledFutureTask.super.run();
    //如果需要周期执行，则在执行完任务以后，设置下一次执行时间
    else if (ScheduledFutureTask.super.runAndReset()) {
        // 计算下次执行该任务的时间
        setNextRunTime();
        //重复执行任务
        reExecutePeriodic(outerTask);
    }
   }

2. 如果当前线程池运行状态不可以执行任务，取消该任务，然后直接返回，否则执行步骤2；
   2. 如果不是周期性任务，调用FutureTask中的run方法执行，会设置执行结果，然后直接返回，否则执行步骤3；
   3. 如果是周期性任务，调用FutureTask中的runAndReset方法执行，不会设置执行结果，然后直接返回，否则执行步骤4和步骤5；
   4. 计算下次执行该任务的具体时间；
   5. 重复执行任务。

   reExecutePeriodic方法

   void reExecutePeriodic(RunnableScheduledFuture<?> task) {
    if (canRunInCurrentRunState(true)) {
        super.getQueue().add(task);
        if (!canRunInCurrentRunState(true) && remove(task))
            task.cancel(false);
        else
            ensurePrestart();
    }
   }

   该方法和delayedExecute方法类似，不同的是：
   1. 由于调用reExecutePeriodic方法时已经执行过一次周期性任务了，所以不会reject当前任务；
   2. 传入的任务一定是周期性任务。

   线程池任务的提交

   首先是schedule方法，该方法是指任务在指定延迟时间到达后触发，只会执行一次。

   public ScheduledFuture<?> schedule(Runnable command,long delay,TimeUnit unit) {
    //参数校验
    if (command == null || unit == null)
        throw new NullPointerException();
    //这里是一个嵌套结构，首先把用户提交的任务包装成ScheduledFutureTask
    //然后在调用decorateTask进行包装，该方法是留给用户去扩展的，默认是个空方法
    RunnableScheduledFuture<?> t = decorateTask(command,
                 new ScheduledFutureTask<Void>(command, null,triggerTime(delay, unit)));
    //包装好任务以后，就进行提交了
    delayedExecute(t);
    return t;
   }

   任务提交方法：

   private void delayedExecute(RunnableScheduledFuture<?> task) {
    //如果线程池已经关闭，则使用拒绝策略把提交任务拒绝掉
    if (isShutdown())
        reject(task);
    else {
        //与ThreadPoolExecutor不同，这里直接把任务加入延迟队列
        super.getQueue().add(task);//使用用的DelayedWorkQueue
        //如果当前状态无法执行任务，则取消
        if (isShutdown() &&!canRunInCurrentRunState(task.isPeriodic()) &&remove(task))
            task.cancel(false);
        else
        //这里是增加一个worker线程，避免提交的任务没有worker去执行
        //原因就是该类没有像ThreadPoolExecutor一样，woker满了才放入队列
            ensurePrestart();
    }
   }

   DelayedWorkQueue

   ScheduledThreadPoolExecutor之所以要自己实现阻塞的工作队列，是因为ScheduledThreadPoolExecutor要求的工作队列有些特殊。
   DelayedWorkQueue是一个基于堆的数据结构，类似于DelayQueue和PriorityQueue。在执行定时任务的时候，每个任务的执行时间都不同，所以DelayedWorkQueue的工作就是按照执行时间的升序来排列，执行时间距离当前时间越近的任务在队列的前面（注意：这里的顺序并不是绝对的，堆中的排序只保证了子节点的下次执行时间要比父节点的下次执行时间要大，而叶子节点之间并不一定是顺序的，下文中会说明）。
   堆结构如下图：
   
   可见，DelayedWorkQueue是一个基于最小堆结构的队列。堆结构可以使用数组表示，可以转换成如下的数组：
   
   在这种结构中，可以发现有如下特性：
   假设，索引值从0开始，子节点的索引值为k，父节点的索引值为p，则：
   1. 一个节点的左子节点的索引为：k = p 2 + 1；
   2. 一个节点的右子节点的索引为：k = (p + 1) 2；
   3. 一个节点的父节点的索引为：p = (k - 1) / 2。

   为什么要使用DelayedWorkQueue呢？

   定时任务执行时需要取出最近要执行的任务，所以任务在队列中每次出队时一定要是当前队列中执行时间最靠前的，所以自然要使用优先级队列。
   DelayedWorkQueue是一个优先级队列，它可以保证每次出队的任务都是当前队列中执行时间最靠前的，由于它是基于堆结构的队列，堆结构在执行插入和删除操作时的最坏时间复杂度是 O(logN)。

   DelayedWorkQueue属性

   // 队列初始容量
   private static final int INITIAL_CAPACITY = 16;
   // 根据初始容量创建RunnableScheduledFuture类型的数组
   private RunnableScheduledFuture<?>[] queue = new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
   private final ReentrantLock lock = new ReentrantLock();
   private int size = 0;
   // leader线程
   private Thread leader = null;
   // 当较新的任务在队列的头部可用时，或者新线程可能需要成为leader，则通过该条件发出信号
   private final Condition available = lock.newCondition();

   注意这里的leader，它是Leader-Follower模式的变体，用于减少不必要的定时等待。什么意思呢？对于多线程的网络模型来说：
   所有线程会有三种身份中的一种：leader和follower，以及一个干活中的状态：proccesser。它的基本原则就是，永远最多只有一个leader。而所有follower都在等待成为leader。线程池启动时会自动产生一个Leader负责等待网络IO事件，当有一个事件产生时，Leader线程首先通知一个Follower线程将其提拔为新的Leader，然后自己就去干活了，去处理这个网络事件，处理完毕后加入Follower线程等待队列，等待下次成为Leader。这种方法可以增强CPU高速缓存相似性，及消除动态内存分配和线程间的数据交换。

   offer方法

   public boolean offer(Runnable x) {
    //参数校验
    if (x == null)
        throw new NullPointerException();
    RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        //查看当前元素数量，如果大于队列长度则进行扩容
        int i = size;
        if (i >= queue.length)
            grow();
        //元素数量加1
        size = i + 1;
        //如果当前队列还没有元素，则直接加入头部
        if (i == 0) {
            queue[0] = e;
            //记录索引
            setIndex(e, 0);
        } else {
            //把任务加入堆中，并调整堆结构，这里就会根据任务的触发时间排列
            //把需要最早执行的任务放在前面
            siftUp(i, e);
        }
        //如果新加入的元素就是队列头，这里有两种情况
        //1.这是用户提交的第一个任务
        //2.新任务进行堆调整以后，排在队列头
        if (queue[0] == e) {
            // leader设置为null为了使在take方法中的线程在通过available.signal();后会执行available.awaitNanos(delay);
            leader = null;
            //加入元素以后，唤醒worker线程
            available.signal();
        }
    } finally {
        lock.unlock();
    }
    return true;
   }

   任务排序sift方法

   private void siftUp(int k, RunnableScheduledFuture<?> key) {
    // 找到父节点的索引
    while (k > 0) {
        // 获取父节点
        int parent = (k ­ 1) >>> 1;
        RunnableScheduledFuture<?> e = queue[parent];
        // 如果key节点的执行时间大于父节点的执行时间，不需要再排序了
        if (key.compareTo(e) >= 0)
            break;
        // 如果key.compareTo(e) < 0，说明key节点的执行时间小于父节点的执行时间，需要把父节点移到后面
        queue[k] = e;
        setIndex(e, k);
        // 设置索引为k
        k = parent;
    }
    // key设置为排序后的位置中
    queue[k] = key;
    setIndex(key, k);
   }

   代码很好理解，就是循环的根据key节点与它的父节点来判断，如果key节点的执行时间小于父节点，则将两个节点交换，使执行时间靠前的节点排列在队列的前面。
   假设新入队的节点的延迟时间（调用getDelay()方法获得）是5，执行过程如下：
   1. 先将新的节点添加到数组的尾部，这时新节点的索引k为7：
   
   2. 计算新父节点的索引：parent = (k - 1) >>> 1，parent = 3，那么queue[3]的时间间隔值为8，因为 5 < 8 ，将执行queue[7] = queue[3]：
   
   3. 这时将k设置为3，继续循环，再次计算parent为1，queue[1]的时间间隔为3，因为 5 > 3 ，这时退出循环，最终k为3：
   
   可见，每次新增节点时，只是根据父节点来判断，而不会影响兄弟节点。

   take方法

   public RunnableScheduledFuture<?> take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        for (;;) {
            RunnableScheduledFuture<?> first = queue[0];
            if (first == null)
                available.await();
            else {
                // 计算当前时间到执行时间的时间间隔
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0)
                    return finishPoll(first);
                first = null; // don't retain ref while waiting
                // leader不为空，阻塞线程
                if (leader != null)
                    available.await();
                else {
                    // leader为空，则把leader设置为当前线程，
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        // 阻塞到执行时间
                        available.awaitNanos(delay);
                    } finally {
                        // 设置leader = null，让其他线程执行
                        available.awaitNanos(delay);
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        // 如果leader不为空，则说明leader的线程正在执行
        available.awaitNanos(delay);
        // 如果queue[0] == null，说明队列为空
        if (leader == null && queue[0] != null)
            available.signal();
        lock.unlock();
    }
   }

   take方法是什么时候调用的呢？在ThreadPoolExecutor中，介绍了getTask方法，工作线程会循环地从workQueue中取任务。但定时任务却不同，因为如果一旦getTask方法取出了任务就开始执行了，而这时可能还没有到执行的时间，所以在take方法中，要保证只有在到指定的执行时间的时候任务才可以被取走。
   再来说一下leader的作用，这里的leader是为了减少不必要的定时等待，当一个线程成为leader时，它只等待下一个节点的时间间隔，但其它线程无限期等待。 leader线程必须在从take（）或poll（）返回之前signal其它线程，除非其他线程成为了leader。
   举例来说，如果没有leader，那么在执行take时，都要执行available.awaitNanos(delay)，假设当前线程执行了该段代码，这时还没有signal，第二个线程也执行了该段代码，则第二个线程也要被阻塞。多个这时执行该段代码是没有作用的，因为只能有一个线程会从take中返回queue[0]（因为有lock），其他线程这时再返回for循环执行时取的queue[0]，已经不是之前的queue[0]了，然后又要继续阻塞。所以，为了不让多个线程频繁的做无用的定时等待，这里增加了leader，如果leader不为空，则说明队列中第一个节点已经在等待出队，这时其它的线程会一直阻塞，减少了无用的阻塞（注意，在finally中调用了signal()来唤醒一个线程，而不是signalAll()）。

   poll 方法

   下面看下poll方法，与take类似，但这里要提供超时功能：

public RunnableScheduledFuture<?> poll(long timeout, TimeUnit unit) throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        for (;;) {
            RunnableScheduledFuture<?> first = queue[0];
            if (first == null) {
                if (nanos <= 0)
                    return null;
                else
                    nanos = available.awaitNanos(nanos);
            } else {
                long delay = first.getDelay(NANOSECONDS);
                // 如果delay <= 0，说明已经到了任务执行的时间，返回
                if (delay <= 0)
                    return finishPoll(first);
                // 如果nanos <= 0，说明已经超时，返回null
                if (nanos <= 0)
                    return null;
                first = null; // don't retain ref while waiting
                // nanos < delay 说明需要等待的时间小于任务要执行的延迟时间
                // leader != null 说明有其它线程正在对任务进行阻塞
                // 这时阻塞当前线程nanos纳秒
                if (nanos < delay || leader != null)
                    nanos = available.awaitNanos(nanos);
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        // 这里的timeLeft表示delay减去实际的等待时间
                        long timeLeft =
                        available.awaitNanos(delay);
                        // 计算剩余的等待时间
                        nanos ­= delay ­ timeLeft;
                    } finally {
                        if (leader == thisThread)
                        leader = null;
                    }
                }
            }
        }
    } finally {
        if (leader == null && queue[0] != null)
            available.signal();
        lock.unlock();
    }
}

并发编程之定时任务&定时线程池.pdf