ThreadPoolExecutor之四：jdk实现的ScheduledThreadPoolExecutor

一、定时任务调度方式常见的有：

1、cron是一个linux下的定时执行工具，一些重要的任务的定时执行可以通过cron来实现，例如每天凌晨1点备份数据等。

2、在JAVA WEB开发中，我们也经常需要用到定时执行任务的功能，JDK提供了Timer类与ScheduledThreadPoolExecutor类实现这个定时功能。

3、quartz

4、分布式任务调度平台XXL-JOB(http://www.cnblogs.com/xuxueli/p/5021979.html)

　　Timer有着不少缺陷，如Timer是单线程模式，调度多个周期性任务时，如果某个任务耗时较久就会影响其它任务的调度；如果某个任务出现异常而没有被catch则可能导致唯一的线程死掉而所有任务都不会再被调度。ScheduledThreadPoolExecutor解决了很多Timer存在的缺陷。

二、ScheduledThreadPoolExecutor使用示例

本章节主要讲述ScheduledThreadPoolExecutor，先通过一个示例了解下基本用法：

package com.dxz.concurrent.schedulethreadpool;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ScheduledThreadPoolExecutorTest {
    public static void main(String[] args) {
        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1);
        BusinessTask task = new BusinessTask();
        // 1秒后开始执行任务，以后每隔2秒执行一次
        executorService.scheduleWithFixedDelay(task, 1000, 2000, TimeUnit.MILLISECONDS);
    }

    private static class BusinessTask implements Runnable {
        @Override
        public void run() {
            System.out.println("任务开始...");
            doBusiness();
            System.out.println("任务结束...");
        }
        
        private void doBusiness() {
            /*String a = null;
            System.out.println("a:="+a.toString());*/
        }
    }
}

结果(片段)：

...
任务结束...
任务开始...
任务结束...
任务开始...
任务结束...
任务开始...
...

如果将上面的doBusiness()中的注释去掉，

结果：（一直被卡住了）

任务开始...

调度功能不起作用，原因：

创建并执行一个在给定初始延迟后首次启用的定期操作，随后，在每一次执行终止和下一次执行开始之间都存在给定的延迟。如果任务的任一执行遇到异常，就会取消后续执行。否则，只能通过执行程序的取消或终止方法来终止该任务。这样就需要把doBusiness()方法的所有可能异常捕获，才能保证定时任务继续执行。

三、ScheduledThreadPoolExecutor原理

先来看看ScheduledThreadPoolExecutor的实现模型，它通过继承ThreadPoolExecutor来重用线程池的功能，里面做了几件事情：

• 为线程池设置了一个DelayedWorkQueue，该queue同时具有PriorityQueue（优先级大的元素会放到队首）和DelayQueue（如果队列里第一个元素的getDelay返回值大于0，则take调用会阻塞）的功能
• 将传入的任务封装成ScheduledFutureTask，这个类有两个特点，实现了java.lang.Comparable和java.util.concurrent.Delayed接口，也就是说里面有两个重要的方法：compareTo和getDelay。ScheduledFutureTask里面存储了该任务距离下次调度还需要的时间（使用的是基于System#nanoTime实现的相对时间，不会因为系统时间改变而改变，如距离下次执行还有10秒，不会因为将系统时间调前6秒而变成4秒后执行）。getDelay方法就是返回当前时间（运行getDelay的这个时刻）距离下次调用之间的时间差；compareTo用于比较两个任务的优先关系，距离下次调度间隔较短的优先级高。那么，当有任务丢进上面说到的DelayedWorkQueue时，因为它有DelayQueue（DelayQueue的内部使用PriorityQueue来实现的）的功能，所以新的任务会与队列中已经存在的任务进行排序，距离下次调度间隔短的任务排在前面，也就是说这个队列并不是先进先出的；另外，在调用DelayedWorkQueue的take方法的时候，如果没有元素，会阻塞，如果有元素而第一个元素的getDelay返回值大于0（前面说过已经排好序了，第一个元素的getDelay不会大于后面元素的getDelay返回值），也会一直阻塞。
• ScheduledFutureTask提供了一个run的实现，线程池执行的就是这个run方法。看看run的源码（本文的代码取自hotspot1.5.0_22，jdk后续版本的代码可能已经不一样了，如jdk1.7中使用了自己实现的DelayedWorkQueue，而不再使用PriorityQueue作为存储，不过从外面看它们的行为还是一样的，所以并不影响对ScheduledThreadPoolExecutor调度机制的理解）：

  public void run() {
              boolean periodic = isPeriodic();
              if (!canRunInCurrentRunState(periodic))
                  cancel(false);
              else if (!periodic)//2
                  ScheduledFutureTask.super.run(); 
              else if (ScheduledFutureTask.super.runAndReset()) {
                  setNextRunTime();
                  reExecutePeriodic(outerTask);
              }
          }

        public boolean cancel(boolean mayInterruptIfRunning) {
            boolean cancelled = super.cancel(mayInterruptIfRunning);
            if (cancelled && removeOnCancel && heapIndex >= 0)
                remove(this);
            return cancelled;
        }

综上，可以看到，ScheduledThreadPoolExecutor执行周期性任务的模型就是：调度一次任务，计算并设置该任务下次间隔，将任务放回队列中供线程池执行。这里的队列起了很大的作用，且有一些特点：距离下次调度间隔短的任务总是在队首，队首的任务若距离下次调度的间隔时间大于0就无法从该队列的take()方法中拿到任务。

周期性任务的入口（

ScheduledThreadPoolExecutor.scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)和

ScheduledThreadPoolExecutor.scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)

）与非周期性任务是类似的，它们处理方式不同的地方在于前文说到的ScheduledFutureTask#run()中。

scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (period <= 0)
            throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(period));
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

固定频率的调度方法：scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) 

    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit) {
        if (command == null || unit == null)   //参数校验
            throw new NullPointerException();
        if (delay <= 0)                        //参数校验
            throw new IllegalArgumentException();
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(-delay));
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

接下来看看ScheduledThreadPoolExecutor.schedule(Callable callable, long delay, TimeUnit unit)和ScheduledThreadPoolExecutor.schedule(Runnable command, long delay, TimeUnit unit)这两个非周期性任务的实现方式，先看看它们的源码：

     public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay,
                                           TimeUnit unit) {
        if (callable == null || unit == null)
            throw new NullPointerException();
        RunnableScheduledFuture<V> t = decorateTask(callable,
            new ScheduledFutureTask<V>(callable,
                                       triggerTime(delay, unit)));
        delayedExecute(t);
        return t;
    }

    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        RunnableScheduledFuture<?> t = decorateTask(command,
            new ScheduledFutureTask<Void>(command, null,
                                          triggerTime(delay, unit)));
        delayedExecute(t);
        return t;
    }

delayedExecute()延迟或周期任务的主要执行方法:

 //延迟或周期任务的主要执行方法。如果线程池关闭，则拒绝任务。否则，将任务添加到队列中，并在必要时启动一个线程来运行它。（我们不能启动线程运行任务因为任务（可能）不应该运行。）如果线程池关闭，任务被添加，如果需要取消，由状态和运行后关闭参数删除它。 

 private void delayedExecute(RunnableScheduledFuture<?> task) {
        if (isShutdown())
            reject(task);
        else {
            super.getQueue().add(task);
            if (isShutdown() &&
                !canRunInCurrentRunState(task.isPeriodic()) &&
                remove(task))
                task.cancel(false);
            else
                ensurePrestart();
        }
    }

实现方式也很简单，在创建ScheduledThreadPoolExecutor内部任务（即ScheduledFutureTask）的时候就将调度间隔计算并设置好，如果当前线程数小于设置的核心线程数，就启动一个线程（可能是线程池刚启动里面还没有线程，也可能是里面的线程执行任务时挂掉了。如果线程池中的线程挂掉了而又没有调用这些schedule方法谁去补充挂掉的线程？不用担心，线程池自己会处理的）去监听队列里的任务，然后将任务放到队列里，在任务执行间隔不大于0的时候，线程就可以拿到这个任务并执行。