线程池原理：ScheduledThreadPoolExecutor

太过爱你忘了你带给我的痛 2022-05-14 01:41 221阅读 0赞

ScheduledThreadPoolExecutor用于定时任务，这里的定时意义在于：

1.  指定延时后执行任务。
2.  周期性重复执行任务。

我们接着分析ScheduledThreadPoolExecutor源码，从类声明开始

## 类声明 ##

public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService { 
        //……
    }        
      
       
        1
        2
        3
        4
        5

ScheduledThreadPoolExecutor继承了ThreadPoolExecutor，实现了ScheduledExecutorService。在线程池的基础上，实现了可调度的线程池功能。上一篇文章已经详细介绍了[ThreadPoolExecutor][]，这里我们先看下ScheduledExecutorService的源码：

## ScheduledExecutorService ##

//可调度的执行者服务接口
    public interface ScheduledExecutorService extends ExecutorService { 
    
        //指定时延后调度执行任务
        public ScheduledFuture<?> schedule(Runnable command,
                                           long delay, TimeUnit unit);
    
        //指定时延后调度执行任务
        public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                               long delay, TimeUnit unit);
    
        //指定时延后开始执行任务，以后每隔period的时长再次执行该任务
        public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                      long initialDelay,
                                                      long period,
                                                      TimeUnit unit);
    
        //指定时延后开始执行任务，以后任务执行完成后等待delay时长，再次执行任务
        public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                         long initialDelay,
                                                         long delay,
                                                         TimeUnit unit);
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23

ScheduledExecutorService实现了ExecutorService，并增加若干定时相关的接口。其中schedule方法用于单次调度执行任务。这里主要理解下后面两个方法。

*  scheduleAtFixedRate：该方法在initialDelay时长后第一次执行任务，以后每隔period时长，再次执行任务。注意，period是从任务开始执行算起的。开始执行任务后，定时器每隔period时长检查该任务是否完成，如果完成则再次启动任务，否则等该任务结束后才再次启动任务，看下图示例。
    
    ![这里写图片描述][SouthEast]
 *  scheduleWithFixDelay：该方法在initialDelay时长后第一次执行任务，以后每当任务执行完成后，等待delay时长，再次执行任务，看下图示例。
    
    ![这里写图片描述][SouthEast 1]

## schedule ##

ScheduledThreadPoolExecutor方法实现了ScheduledExecutorService，schedule方法调度的任务只执行一次。

先看下schedule方法的实现：

//delay时长后执行任务command，该任务只执行一次
    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
        if (command == null || unit == null)
          throw new NullPointerException();
        //这里的decorateTask方法仅仅返回第二个参数
        RunnableScheduledFuture<?> t = decorateTask(command,
                                                    new ScheduledFutureTask<Void>(command, null,
                                                                                  triggerTime(delay, unit)));
        //延时或者周期执行任务的主要方法
        delayedExecute(t);
        return t;
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12

我们先屡下ScheduledFuture、RunnableScheduledFuture、ScheduledFutureTask的关系，看下类图：

![这里写图片描述][SouthEast 2]

这个类图比较复杂，其中浅色部分都是我们已经学习过了，深色部分我们之前没有接触过，所以重点学习这几个类，从上往下依次看。

## Delayed接口 ##

Delayed接口提供了getDelay方法，该方法返回对象剩余时延。接口继承了Comparable接口，表示对象支持排序，看下该接口的定义：

//继承Comparable接口，表示该类对象支持排序
    public interface Delayed extends Comparable<Delayed> { 
        //返回该对象剩余时延
        long getDelay(TimeUnit unit);
    }
      
       
        1
        2
        3
        4
        5

Delayed接口很简单，继续看ScheduledFuture接口。

## ScheduledFuture接口 ##

ScheduledFuture是延时的Future，仅仅继承了Delayed和Future接口，并没有添加其他方法，看下该接口的定义：

//仅仅继承了Delayed和Future接口
    public interface ScheduledFuture<V> extends Delayed, Future<V> { 
    }
      
       
        1
        2
        3

## RunnableScheduledFuture接口 ##

可运行的ScheduledFuture，该接口继承了ScheduledFuture和RunnableFuture接口。

public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> { 
        //是否是周期任务，周期任务可被调度运行多次，非周期任务只被运行一次
        boolean isPeriodic();
    }
      
       
        1
        2
        3
        4

## ScheduledFutureTask类 ##

该类是ScheduledThreadPoolExecutor的内部类，继承了FutureTask，实现了RunnableScheduledFuture接口。[FutureTask][]我们在介绍线程池的时候讲过。先看下ScheduledFutureTask的构造方法：

ScheduledFutureTask(Runnable r, V result, long ns, long period) {
        //调用父类FutureTask的构造方法
        super(r, result);
        //time表示任务下次执行的时间
        this.time = ns;
        //周期任务，正数表示按照固定速率，负数表示按照固定时延
        this.period = period;
        //任务的编号
        this.sequenceNumber = sequencer.getAndIncrement();
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10

这里需要注意几点，

1.  time表示任务下一次执行的时间，单位为纳秒。
2.  period=0表示该任务不是周期性任务，正数表示每隔period时长执行任务，负数表示任务执行完成后到下一次被调度运行的延时时间。
3.  sequenceNumber表示该任务的编号，通过线程池的sequencer成员变量从0开始生成编号。

继续看下getDelay方法：

### getDelay ###

//实现Delayed接口的getDelay方法，返回任务开始执行的剩余时间
    public long getDelay(TimeUnit unit) {
        return unit.convert(time - now(), TimeUnit.NANOSECONDS);
    }
      
       
        1
        2
        3
        4

这个方法其实就是任务开始执行的倒计时时间，通过任务预期执行时间减去当前时间获得，单位是纳秒。

### compareTo ###

该方法实现了Comparable接口的compareTo方法，比较两个任务的”大小”。后面我们会讲到，可调度的线程池其实利用了可排序的延时队列，延时队列保存了ScheduledFutureTask任务，并且队列中的元素会根据开始执行的倒计时时间排序，剩余等待时间最少的将会被最先调度运行。这里排序策略就是根据compareTo方法实现的。

public int compareTo(Delayed other) {
        if (other == this)
          return 0;
        if (other instanceof ScheduledFutureTask) {
          ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
          long diff = time - x.time;
          //小于0，说明当前任务的执行时间点早于other，要排在延时队列other的前面
          if (diff < 0)
            return -1;
          //大于0，说明当前任务的执行时间点晚于other，要排在延时队列other的后面
          else if (diff > 0)
            return 1;
          //如果两个任务的执行时间点一样，比较两个任务的编号，编号小的排在队列前面，编号大的排在队列后面
          else if (sequenceNumber < x.sequenceNumber)
            return -1;
          else
            return 1;
        }
        //如果任务类型不是ScheduledFutureTask，通过getDelay方法比较
        long d = (getDelay(TimeUnit.NANOSECONDS) -
                  other.getDelay(TimeUnit.NANOSECONDS));
        return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23

### setNextRunTime ###

任务执行完后，设置下次执行的时间

private void setNextRunTime() {
        long p = period;
        //p>0，说明是固定速率运行的任务，在原来任务开始执行时间的基础上加上p即可
        if (p > 0)
          time += p;
        //p<0，说明是固定时延运行的任务，下次执行时间在当前时间(任务执行完成的时间)的基础上加上-p的时间
        else
          time = triggerTime(-p);
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9

任务执行完成后需要确定下次执行的时间，如果任务是以固定速率运行的，下次开始执行时间就是上次任务开始执行时间加上period。如果任务是以固定延时执行的，下次开始执行时间就是当前时间（上次任务线束时间）加上period（取正值）。

### cancel ###

取消任务的执行，重点关注将取消的任务从队列移除的逻辑。

public boolean cancel(boolean mayInterruptIfRunning) {
        //调用FutureTask的cancel方法
        boolean cancelled = super.cancel(mayInterruptIfRunning);
        //cancelled: 任务取消成功
        //removeOnCancel：任务取消后从队列移除
        //headIndex：任务原先处于二叉堆的位置
        if (cancelled && removeOnCancel && heapIndex >= 0)
          //从队列中移除，该方法是ThreadPoolExecutor的方法
          remove(this);
        //返回是否取消成功
        return cancelled;
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12

### run ###

ScheduledFutureTask重写了FutureTask的run方法。

public void run() {
        boolean periodic = isPeriodic();
        //如果当前状态下不能执行任务，则取消任务
        if (!canRunInCurrentRunState(periodic))
          cancel(false);
        //不是周期性任务，执行一次任务即可，调用父类的run方法
        else if (!periodic)
          ScheduledFutureTask.super.run();
        //是周期性任务，调用FutureTask的runAndReset方法，方法执行完成后
        //重新设置任务下一次执行的时间，并将该任务重新入队，等待再次被调度
        else if (ScheduledFutureTask.super.runAndReset()) {
          setNextRunTime();
          reExecutePeriodic(outerTask);
        }
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15

注释已经解释的很清楚了，重点看下FutureTask的runAndReset方法，该方法是为任务多次执行而设计的。runAndReset方法执行完任务后不会设置任务的执行结果，也不会去更新任务的状态，维持任务的状态为初始状态（NEW状态），这也是该方法和FutureTask的run方法的区别。

好了，讲完了ScheduledFutureTask，接着看ScheduledPoolExecutor源码。

通常我们通过submit或者execute方法将任务提交给线程池执行，这两个方法最终都是调用了schedule方法，前面已经讲过，schedule方法只会调度任务执行一次。那么ScheduledThreadPoolExecutor是怎样调度固定周期或延时的任务的呢？是通过scheduledAtFixedRate和scheduledAtFixedDelay方法实现的，我们先看下scheduledAtFixedRate源码：

## scheduledAtFixedRate ##

关于该方法的说明，我们在ScheduledExecutorService接口已经说明过了，这里主要看下实现。

//注意，固定速率和固定时延，传入的参数都是Runnable，也就是说这种定时任务是没有返回值的
    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);
        //outerTask表示将会重新入队的任务
        sft.outerTask = t;
        //稍后说明
        delayedExecute(t);
        return t;
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22

其实主要创建了一个带有初始延时和固定周期的任务，类似的，scheduledAtFixedDelay创建一个带有初始延时和任务间固定延时的任务。

## scheduledAtFixedDelay ##

和scheduledAtFixedRate类似，唯一不同的地方在于在于创建的ScheduledFutureTask不同，FixedRate和FixedDelay也是通过ScheduledFutureTask体现的。这里不再展示代码了。

## delayedExecute ##

前面讲到的schedule、scheduleAtFixedRate和scheduleAtFixedDelay最后都调用了delayedExecute方法，该方法是定时任务执行的主要方法。看下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
            //增加一个worker，就算corePoolSize=0也要增加一个worker
            ensurePrestart();
        }
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17

delayedExecute方法的逻辑也很简单，主要就是将任务添加到等待队列并增加一个worker，增加的worker并不能立即执行该任务，因为该任务可能要等待一定时间后才能执行。

对于ScheduledThreadPoolExecutor，worker添加到线程池后会在等待队列上等待获取任务，这点是和ThreadPoolExecutor一致的。但是worker是怎么从等待队列取定时任务的？该等待队列队首应该保存的是最近将要执行的任务，如果队首任务的开始执行时间还未到，worker也应该继续等待。

ScheduledThreadPoolExecutor实现了一个延时队列，该队列不仅实现了阻塞队列的功能，也实现了排序功能。后面我们会发现，该队列是通过二叉堆实现的，理解了该队列基本上能够理解ScheduledThreadPoolExecutor了，因此我们好好学习下该队列。

ScheduledThreadPoolExecutor内部类DelayedWorkQueue就是保存定时任务的等待队列。

## DelayedWorkQueue ##

看下DelayedWorkQueue的声明：

static class DelayedWorkQueue extends AbstractQueue<Runnable>
            implements BlockingQueue<Runnable> {
            //……
    }
      
       
        1
        2
        3
        4

DelayedWorkQueue继承了AbstractQueue抽象类、实现了BlockingQueue接口。

理解DelayedWorkQueue之前需要理解[堆排序][Link 1]，这里的堆排序算法和DelayedWorkQueue的稍有不同，但是基本思想是相同的。

堆排序是通过数组实现的，因此DelayedWorkQueue定义了一个数组作为等待队列。

//队列初始容量
    private static final int INITIAL_CAPACITY = 16;
    //数组用来存储定时任务，通过数组实现堆排序
    private RunnableScheduledFuture[] queue = new RunnableScheduledFuture[INITIAL_CAPACITY];
      
       
        1
        2
        3
        4

DelayedWorkQueue保存了当前在队首等待的线程：

private Thread leader = null;
      
       
        1

当一个线程成为leader，它只要等待队首任务的delay时间即可，其他线程会无条件等待。leader取到任务返回前要通知其他线程，直到有线程成为新的leader。每当队首的定时任务被其他更早需要执行的任务替换时，leader设置为null，其他等待的线程（被当前leader通知）和当前的leader重新竞争成为leader。

DelayedWorkQueue定义了锁lock和条件available用于线程竞争成为leader。

private final ReentrantLock lock = new ReentrantLock();
    private final Condition available = lock.newCondition();
      
       
        1
        2

当一个新的任务成为队首，或者需要有新的线程成为leader时，available条件将会被通知。

线程取任务时需要在available条件上等待，当被通知时，该线程可能会成为新的leader。

我们先看下DelayedWorkQueue的take方法

### take ###

public RunnableScheduledFuture take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
          for (;;) {
            //取堆顶的任务，堆顶是最近要执行的任务
            RunnableScheduledFuture first = queue[0];
            //堆顶为空，线程要在条件available上等待
            if (first == null)
              available.await();
            else {
              //堆顶任务还要多长时间才能执行
              long delay = first.getDelay(TimeUnit.NANOSECONDS);
              //堆顶任务已经可以执行了，finishPoll会重新调整堆，使其满足最小堆特性，该方法设置任务在
              //堆中的index为-1并返回该任务
              if (delay <= 0)
                return finishPoll(first);
              //如果leader不为空，说明已经有线程成为leader并等待堆顶任务
              //到达执行时间，此时，其他线程都需要在available条件上等待
              else if (leader != null)
                available.await();
              else {
                //leader为空，当前线程成为新的leader
                Thread thisThread = Thread.currentThread();
                leader = thisThread;
                try {
                  //当前线程已经成为leader了，只需要等待堆顶任务到达执行时间即可
                  available.awaitNanos(delay);
                } finally {
                  //返回堆顶元素之前将leader设置为空
                  if (leader == thisThread)
                    leader = null;
                }
              }
            }
          }
        } finally {
          //通知其他在available条件等待的线程，这些线程可以去竞争成为新的leader
          if (leader == null && queue[0] != null)
            available.signal();
          lock.unlock();
        }
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23
        24
        25
        26
        27
        28
        29
        30
        31
        32
        33
        34
        35
        36
        37
        38
        39
        40
        41
        42
        43

再梳理下take方法的逻辑

*  如果堆顶元素为空，在available条件上等待。
 *  如果堆顶任务的执行时间已到，将堆顶元素替换为堆的最后一个元素并调整堆使其满足最小堆特性，同时设置任务在堆中索引为-1，返回该任务。
 *  如果leader不为空，说明已经有线程成为leader了，其他线程都要在available条件上等待。
 *  如果leader为空，当前线程成为新的leader，并等待直到堆顶任务执行时间到达。
 *  take方法返回之前，将leader设置为空，并通知其他线程。

继续看下finishPool方法：

private RunnableScheduledFuture finishPoll(RunnableScheduledFuture f) {
        //堆元素数量减1
        int s = --size;
        //取堆的最后一个元素
        RunnableScheduledFuture x = queue[s];
        queue[s] = null;
        if (s != 0)
          //调整堆，使其重新满足最小堆特性，从位置0开始往堆的底层调整
          siftDown(0, x);
        //该任务在堆中的索引设置为-1
        setIndex(f, -1);
        //返回该任务
        return f;
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14

### 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;
          //队列元素已经大于等于数组的长度，需要扩容，新堆的容易是原来堆容量的1.5倍
          if (i >= queue.length)
            grow();
          //堆中元素增加1
          size = i + 1;
          //调整堆
          if (i == 0) {
            queue[0] = e;
            setIndex(e, 0);
          } else {
            siftUp(i, e);
          }
          if (queue[0] == e) {
            leader = null;
            //通知其他在available条件上等待的线程，这些线程可以竞争成为新的leader
            available.signal();
          }
        } finally {
          lock.unlock();
        }
        return true;
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23
        24
        25
        26
        27
        28
        29
        30

### siftUp ###

该方法是调整堆的方法，调整堆的目的是使其满足最小堆的特性。

//从位置k开始往堆顶方向查找，直到找到key保存的位置
    private void siftUp(int k, RunnableScheduledFuture key) {
        while (k > 0) {
          //parent是父节点的索引
          int parent = (k - 1) >>> 1;
          RunnableScheduledFuture e = queue[parent];
          //如果父节点比子节点e的执行时间要早，说明已经符合最小堆的特性，跳出循环
          if (key.compareTo(e) >= 0)
            break;
          //子节点比父节点更早执行，将子节点位置的值替换为父节点
          queue[k] = e;
          setIndex(e, k);
          //继续往上查找
          k = parent;
        }
        //k是最终key存放的位置
        queue[k] = key;
        setIndex(key, k);
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19

看下siftUp示例图，对于左边这个堆来说，在位置K处往堆顶方向查找key=12的位置，因为父节点值为23，大于12，因此将23移到位置K处，位置K上移到父节点所在位置，继续往堆顶方向查找key=12的位置。

如果查找key=50，因为父节点23小于50，因此位置K就是key=50的最终保存位置。

![这里写图片描述][SouthEast 3]

### siftDown ###

\*\*重点内容\*\*

该方法和siftUp方法类似

//从位置k处开始往下查找，找到key的保存位置
    private void siftDown(int k, RunnableScheduledFuture key) {
        //从half开始，就不再有孩子节点的，这是一个优化
        int half = size >>> 1;
        while (k < half) {
          //左孩子位置
          int child = (k << 1) + 1;
          RunnableScheduledFuture c = queue[child];
          //右孩子位置
          int right = child + 1;
          //如果右孩子存在，并且右孩子比左孩子更早执行，更新c为右孩子
          if (right < size && c.compareTo(queue[right]) > 0)
            c = queue[child = right];
          //以上做的都是取两个孩子中更早执行的那个孩子节点，取到后和key比较
          //如果key比两个孩子都更早执行，位置k就是key的最终位置了，跳出循环
          if (key.compareTo(c) <= 0)
            break;
          //更早执行的孩子放到父节点处
          queue[k] = c;
          setIndex(c, k);
          //继续往下查找
          k = child;
        }
        queue[k] = key;
        setIndex(key, k);
    }
      
       
        1
        2
        3
        4
        5
        6
        7
        8
        9
        10
        11
        12
        13
        14
        15
        16
        17
        18
        19
        20
        21
        22
        23
        24
        25
        26

看下siftDown的示例图，对于左边的堆来说，在位置K处开始往下查找key的位置，如果key=12，因为12小于两个孩子中的最小结点35，因此位置K就是key=12的最终保存位置。如果key=50，因为50大于两个孩子结点中的最小结点35，因此将35上移到父节点，位置K下移到35所在的位置，继续往堆底查找。

![这里写图片描述][SouthEast 4]

[ThreadPoolExecutor]: http://blog.csdn.net/luanmousheng/article/details/77688356
[SouthEast]: /images/20220514/546cd1251bef48d487bd99f4775c786f.png
[SouthEast 1]: /images/20220514/fd65a3bc87ba43f2bea2b8579b097d5c.png
[SouthEast 2]: /images/20220514/a961fd1ac199472f826e6fc6822880e7.png
[FutureTask]: http://blog.csdn.net/luanmousheng/article/details/77417776
[Link 1]: http://blog.csdn.net/luanmousheng/article/details/77108228
[SouthEast 3]: /images/20220514/ffcf58f93ecc44f2b11dabac402fa92a.png
[SouthEast 4]: /images/20220514/3b56414a2d1a4d1a97b527746207adc5.png