Android小知识-定时任务ScheduledThreadPoolExecutor-白红宇

本平台的文章更新会有延迟，大家可以关注微信公众号-顾林海，包括年底前会更新kotlin由浅入深系列教程，目前计划在微信公众号进行首发，如果大家想获取最新教程，请关注微信公众号，谢谢!

ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，而ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务，ScheduledThreadPoolExecutor是一个实现定时任务的类，可以在给定的延迟后运行命令，或者定期执行命令。

ScheduledThreadPoolExecutor定义了四个构造函数，这四个构造函数如下：

/*** @param corePoolSize 核心线程池的大小*/public ScheduledThreadPoolExecutor(int corePoolSize) {    super(corePoolSize, Integer.MAX_VALUE,            DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,            new DelayedWorkQueue());}/*** @param corePoolSize  核心线程池的大小* @param threadFactory 用于设置创建线程的工厂*/public ScheduledThreadPoolExecutor(int corePoolSize,                                   ThreadFactory threadFactory) {    super(corePoolSize, Integer.MAX_VALUE,            DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,            new DelayedWorkQueue(), threadFactory);}/*** @param corePoolSize 核心线程池的大小* @param handler      饱和策略*/public ScheduledThreadPoolExecutor(int corePoolSize,                                   RejectedExecutionHandler handler) {    super(corePoolSize, Integer.MAX_VALUE,            DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,            new DelayedWorkQueue(), handler);}/*** @param corePoolSize  核心线程池的大小* @param threadFactory 用于设置创建线程的工厂* @param handler       饱和策略*/public ScheduledThreadPoolExecutor(int corePoolSize,                                   ThreadFactory threadFactory,                                   RejectedExecutionHandler handler) {    super(corePoolSize, Integer.MAX_VALUE,            DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,            new DelayedWorkQueue(), threadFactory, handler);}复制代码

通过源码可以发现，ScheduledThreadPoolExecutor的构造器都是调用父类的构造器也就是ThreadPoolExecutor的构造器，以此来创建一个线程池。

ThreadPoolExecutor的构造器如下：

public ThreadPoolExecutor(int corePoolSize,                          int maximumPoolSize,                          long keepAliveTime,                          TimeUnit unit,                          BlockingQueue
    
      workQueue) {    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,            Executors.defaultThreadFactory(), defaultHandler);}public ThreadPoolExecutor(int corePoolSize,                          int maximumPoolSize,                          long keepAliveTime,                          TimeUnit unit,                          BlockingQueue
     
       workQueue,                          ThreadFactory threadFactory) {    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,            threadFactory, defaultHandler);}public ThreadPoolExecutor(int corePoolSize,                          int maximumPoolSize,                          long keepAliveTime,                          TimeUnit unit,                          BlockingQueue
      
        workQueue,                          RejectedExecutionHandler handler) {    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,            Executors.defaultThreadFactory(), handler);}public ThreadPoolExecutor(int corePoolSize,                          int maximumPoolSize,                          long keepAliveTime,                          TimeUnit unit,                          BlockingQueue
       
         workQueue,                          ThreadFactory threadFactory,                          RejectedExecutionHandler handler) {    if (corePoolSize < 0 ||            maximumPoolSize <= 0 ||            maximumPoolSize < corePoolSize ||            keepAliveTime < 0)        throw new IllegalArgumentException();    if (workQueue == null || threadFactory == null || handler == null)        throw new NullPointerException();    this.corePoolSize = corePoolSize;    this.maximumPoolSize = maximumPoolSize;    this.workQueue = workQueue;    this.keepAliveTime = unit.toNanos(keepAliveTime);    this.threadFactory = threadFactory;    this.handler = handler;}复制代码

创建一个线程池时需要输入几个参数，如下：

corePoolSize（线程池的基本大小）：当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其它空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建，会把到达的任务放到缓存队列当中。如果调用了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有基本线程，或调用线程池的prestartCoreThread()方法，线程池会提前创建一个线程。

maximumPoolSize（线程池最大数量）：线程池允许创建的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是，如果使用了无界的任务队列这个参数就没什么效果。

KeepAliveTime（线程活动保持时间）：线程池的工作线程空闲后，保持存货的时间。如果任务很多，并且每个任务执行的时间比较短，可以调大时间，提供线程的利用率。

unit（线程活动保持时间的单位）：可选的单位有天（DAYS）、小时（HOURS）、分钟（MINUTES）、毫秒（MILLISECONDS）、微妙（MICROSECONDS）、千分之一毫秒和纳秒（NANOSECONDS、千分之一微妙）。

workQueue（任务队列）：用于保持等待执行的任务的阻塞队列，可以选择以下几个阻塞队列。（1）ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按FIFO（先进先出）原则对元素进行排序。（2）LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO排序元素，吞吐量通常要高于ArrayBlockingQueue。（3）SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue。（4）PriorityBlockingQueue：一个具有优先级的无限阻塞队列。

ThreadFactory：用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字。

RejectedExecutionHandler（饱和策略）：当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。在JDK1.5中Java线程池框架提供了4种策略（也可通过实现RejectedExecutionHandler接口自定义策略）。（1）AbortPolicy：直接抛出异常。（2）CallerRunsPolicy：只用调用者所在线程来运行任务。（3）DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。（4）DiscardPolicy：处理，丢弃掉。

在ScheduledThreadPoolExecutor构造器中使用了工作队列java.util.concurrent.ScheduledThreadPoolExecutor.DelayedWorkQueue，DelayedWorkQueue是一个无界的BlockingQueue，用于放置实现了Delayed接口的对象，其中的对象只能在其到期才能从队列中取走。

由于ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，因此它也实现了ThreadPoolExecutor的方法，如下：

public void execute(Runnable command) {        ...    }    public Future
     submit(Runnable task) {        ...    }    public 
    
      Future
     
       submit(Runnable task, T result) {        ...    }    public 
      
        Future
       
         submit(Callable
        
          task) {        ...    }复制代码

同时它也有自己的定时执行任务的方法：

/**  * 延迟delay时间后开始执行task，无法获取task的执行结果。  */ public ScheduledFuture
     schedule(Runnable command,                                       long delay,                                       TimeUnit unit) {     ... } /**  * 延迟delay时间后开始执行callable，它接收的是一个Callable实例，  * 此方法会返回一个ScheduleFuture对象，通过ScheduleFuture我们  * 可以取消一个未执行的task，也可以获得这个task的执行结果。  */ public 
    
      ScheduledFuture
     
       schedule(Callable
      
        callable,                                           long delay,                                           TimeUnit unit) {    ... } /**  * 延迟initialDelay时间后开始执行command，并且按照period时间周期性  * 重复调用，当任务执行时间大于间隔时间时，之后的任务都会延迟，此时与  * Timer中的schedule方法类似。  */ public ScheduledFuture
        scheduleAtFixedRate(Runnable command,                                                  long initialDelay,                                                  long period,                                                  TimeUnit unit) {    ... }/** *延迟initialDelay时间后开始执行command，并且按照period时间周期性重复 *调用，这里的间隔时间delay是等上一个任务完全执行完毕才开始计算。 */ public ScheduledFuture
        scheduleWithFixedDelay(Runnable command,                                                     long initialDelay,                                                     long delay,                                                     TimeUnit unit) {    ... }复制代码

ScheduledThreadPoolExecutor把待调度的任务放到一个DelayedWorkQueue ，并且DelayedWorkQueue 是一个无界队列，ThreadPoolExecutor的maximumPoolSize在ScheduledThreadPoolExecutor中没有什么意义。整个ScheduledThreadPoolExecutor的运行可以分为两大部分。

1、当调用ScheduledThreadPoolExecutor的上面4个方法时，会向ScheduledThreadPoolExecutor的DelayedWorkQueue 添加一个实现了RunnableScheduleFuture接口的ScheduledFutureTask，如下ScheduledThreadPoolExecutor其中的一个schedule方法。

public 
    
      ScheduledFuture
     
       schedule(Callable
      
        callable,                                       long delay,                                       TimeUnit unit) {    if (callable == null || unit == null)        throw new NullPointerException();    RunnableScheduledFuture
       
         t = decorateTask(callable,        new ScheduledFutureTask
        
         (callable,                                   triggerTime(delay, unit),                                   sequencer.getAndIncrement()));    delayedExecute(t);    return t;}private void delayedExecute(RunnableScheduledFuture
          task) {    if (isShutdown())        reject(task);    else {        super.getQueue().add(task);//向ScheduledThreadPoolExecutor的DelayedWorkQueue添加一个实现了RunnableScheduleFuture接口的ScheduledFutureTask        if (isShutdown() &&            !canRunInCurrentRunState(task.isPeriodic()) &&            remove(task))            task.cancel(false);        else            ensurePrestart();    }}复制代码

2、线程池中的线程从DelayedWorkQueue 中获取ScheduledFutureTask，然后执行。

ScheduledFutureTask是ScheduledThreadPoolExecutor的内部类并继承自FutureTask，包含3个成员变量。

//ong型成员变量sequenceNumber，表示这个任务被添加到//ScheduledThreadPoolExecutor中的序号。private final long sequenceNumber;//long型成员变量time，表示这个任务将要被执行的具体时间。private volatile long time;//long型成员变量period，表示任务执行的间隔周期。private final long period;ScheduledFutureTask内部实现了compareTo()方法，用于对task的排序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;}复制代码

排序时，time小的排在前面，如果两个ScheduledFutureTask的time相同，就比较sequenceNumber，sequenceNumber小的排在前面。

DelayedWorkQueue 内部使用了二叉堆算法，DelayedWorkQueue 中的元素第一个元素永远是延迟时间最小的那个元素。当执行 schedule 方法是。如果不是重复的任务，那任务从 DelayedWorkQueue 取出之后执行完了就结束了。如果是重复的任务，那在执行结束前会重置执行时间并将自己重新加入到 DelayedWorkQueue 中

总结来说，ScheduledThreadPoolExecutor是一个实现ScheduledExecutorService的可以调度任务的执行框架；DelayedWorkQueue是一个数组实现的阻塞队列，根据任务所提供的时间参数来调整位置，实际上就是个小根堆（优先队列）；ScheduledFutureTask包含任务单元，存有时间、周期、外部任务、堆下标等调度过程中必须用到的参数，被工作线程执行。ScheduledThreadPoolExecutor与Timer都是用作定时任务，它们直接的差异是Timer使用的是绝对时间，系统时间的改变会对Timer产生一定的影响；而ScheduledThreadPoolExecutor使用的是相对时间，不会导致这个问题。Timer使用的是单线程来处理任务，长时间运行的任务会导致其他任务的延迟处理；而ScheduledThreadPoolExecutor可以自定义线程数量。并且Timer没有对运行时异常进行处理，一旦某个任务触发运行时异常，会导致整个Timer崩溃；而ScheduledThreadPoolExecutor对运行时异常做了捕获（通过afterExecute()回调方法中进行处理），所以更安全。