ScheduledThreadPoolExecutor中定时周期任务的实现源码分析
松饼人 人气:2ScheduledThreadPoolExecutor是一个定时任务线程池，相比于ThreadPoolExecutor最大的不同在于其阻塞队列的实现
首先看一下其构造方法：
1 public ScheduledThreadPoolExecutor(int corePoolSize,
2 ThreadFactory threadFactory,
3 RejectedExecutionHandler handler) {
4 super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
5 new DelayedWorkQueue(), threadFactory, handler);
6 }
ScheduledThreadPoolExecutor是继承自ThreadPoolExecutor的，可以看到这里实际上调用了ThreadPoolExecutor的构造方法，而最大的不同在于这里使用了默认的DelayedWorkQueue“阻塞队列”，这是后续能够实现定时任务的关键
在ScheduledThreadPoolExecutor中使用scheduleWithFixedDelay或者scheduleAtFixedRate方法来完成定时周期任务
以scheduleWithFixedDelay为例
scheduleWithFixedDelay方法：
1 public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
2 long initialDelay,
3 long delay,
4 TimeUnit unit) {
5 if (command == null || unit == null)
6 throw new NullPointerException();
7 if (delay <= 0)
8 throw new IllegalArgumentException();
9 ScheduledFutureTask<Void> sft =
10 new ScheduledFutureTask<Void>(command,
11 null,
12 triggerTime(initialDelay, unit),
13 unit.toNanos(-delay));
14 RunnableScheduledFuture<Void> t = decorateTask(command, sft);
15 sft.outerTask = t;
16 delayedExecute(t);
17 return t;
18 }
这里首先会将我们的任务包装成ScheduledFutureTask
（这里的delay在传入ScheduledFutureTask的构造方法时变为了负的，这是和scheduleAtFixedRate方法唯一不一样的地方）
ScheduledFutureTask方法：
1 private void delayedExecute(RunnableScheduledFuture<?> task) {
2 if (isShutdown())
3 reject(task);
4 else {
5 super.getQueue().add(task);
6 if (isShutdown() &&
7 !canRunInCurrentRunState(task.isPeriodic()) &&
8 remove(task))
9 task.cancel(false);
10 else
11 ensurePrestart();
12 }
13 }
这里不同于ThreadPoolExecutor中的处理，并没有考虑coreSize和maxSize和任务之间的关系，而是直接将任务提交到阻塞队列DelayedWorkQueue中
DelayedWorkQueue的add方法：
1 public boolean add(Runnable e) {
2 return offer(e);
3 }
4
5 public boolean offer(Runnable x) {
6 if (x == null)
7 throw new NullPointerException();
8 RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
9 final ReentrantLock lock = this.lock;
10 lock.lock();
11 try {
12 int i = size;
13 if (i >= queue.length)
14 grow();
15 size = i + 1;
16 if (i == 0) {
17 queue[0] = e;
18 setIndex(e, 0);
19 } else {
20 siftUp(i, e);
21 }
22 if (queue[0] == e) {
23 leader = null;
24 available.signal();
25 }
26 } finally {
27 lock.unlock();
28 }
29 return true;
30 }
实际上调用了offer方法，从这里就可以看出这个“阻塞队列”的不同之处
DelayedWorkQueue中有这些成员：
1 private static final int INITIAL_CAPACITY = 16; 2 private RunnableScheduledFuture<?>[] queue = 3 new RunnableScheduledFuture<?>[INITIAL_CAPACITY]; 4 private int size = 0; 5 private Thread leader = null;
在DelayedWorkQueue内部维护的是queue这个初始大小16的数组，其实就是一个小根堆
回到offer方法
由于是在多线程环境，这里的操作使用了重入锁保证原子性
若是在size大于数组的长度情况下，就需要调用grow方法来扩容
grow方法：
1 private void grow() {
2 int oldCapacity = queue.length;
3 int newCapacity = oldCapacity + (oldCapacity >> 1); // grow 50%
4 if (newCapacity < 0) // overflow
5 newCapacity = Integer.MAX_VALUE;
6 queue = Arrays.copyOf(queue, newCapacity);
7 }
可以看到这是一个非常简单的扩容机制，申请一个1.5倍大小的新数组，再将原来的数据copy上去
回到offer方法，在调整完容量后，就需要进行数据的插入，使其形成一个小根堆
可以看到，在if-else判断中，首先检查是不是第一个元素，若是第一个，则直接放入数组，同时调用
setIndex方法，和任务关联
setIndex方法：
1 private void setIndex(RunnableScheduledFuture<?> f, int idx) {
2 if (f instanceof ScheduledFutureTask)
3 ((ScheduledFutureTask)f).heapIndex = idx;
4 }
这个方法很简单，将下标关联到之前包装好的任务ScheduledFutureTask中
若不是第一个元素，则需要调用siftUp，进行小根堆的调整
siftUp方法：
1 private void siftUp(int k, RunnableScheduledFuture<?> key) {
2 while (k > 0) {
3 int parent = (k - 1) >>> 1;
4 RunnableScheduledFuture<?> e = queue[parent];
5 if (key.compareTo(e) >= 0)
6 break;
7 queue[k] = e;
8 setIndex(e, k);
9 k = parent;
10 }
11 queue[k] = key;
12 setIndex(key, k);
13 }
因为小根堆实际上就是一个二叉树，利用二叉树的性质根据当前要插入节点的下标，得到其父节点的下标parent ，再和父节点的RunnableScheduledFuture对象进行compareTo的比较（RunnableScheduledFuture继承了Comparable接口）
compareTo的实现：
1 public int compareTo(Delayed other) {
2 if (other == this) // compare zero if same object
3 return 0;
4 if (other instanceof ScheduledFutureTask) {
5 ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
6 long diff = time - x.time;
7 if (diff < 0)
8 return -1;
9 else if (diff > 0)
10 return 1;
11 else if (sequenceNumber < x.sequenceNumber)
12 return -1;
13 else
14 return 1;
15 }
16 long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
17 return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
18 }
这里的逻辑比较简单，只需要看第二个if
在前面ScheduledFutureTask包装我们的任务的时候，其构造方法如下：
1 ScheduledFutureTask(Runnable r, V result, long ns, long period) {
2 super(r, result);
3 this.time = ns;
4 this.period = period;
5 this.sequenceNumber = sequencer.getAndIncrement();
6 }
这里的time 也就是initialDelay，period 就是-delay，sequenceNumber 是一个全局自增的序列号
那么在上面的compareTo方法中，就首先根据子节点的initialDelay和父节点的initialDelay比较
若是子节点小于父节点，返回-1，子节点大于父节点返回1
若是相等，则根据序列号比较，序列号小的返回-1
回到siftUp方法，通过compareTo方法，若是大于等于0，就说明子节点大于父节点，不需要做调整，结束循环
若是小于0，说明子节点小于父节点，那么就需要将父节点先交换到当前位置，再将k变成parent，在下一次循环时，就会找parent的parent，重复上述操作，直至构成小根堆
最后将要插入的节点放入queue中合适的位置
那么在后续的任务添加中，就会根据任务的initialDelay，以及创建时间，构建一个小根堆
回到offer方法，在小根堆中插入完节点后，若是第一次插入， 将leader（Thread对象）置为null，利用available（Condition对象）唤醒Lock 的AQS上的阻塞
DelayedWorkQueue的add到此结束，回到delayedExecute方法中，在完成向阻塞队列添加任务后，发现线程池中并没有一个worker在工作，接下来的工作就由ThreadPoolExecutor的ensurePrestart方法实现：
1 void ensurePrestart() {
2 int wc = workerCountOf(ctl.get());
3 if (wc < corePoolSize)
4 addWorker(null, true);
5 else if (wc == 0)
6 addWorker(null, false);
7 }
可以看到这里根据ctl的取值，与corePoolSize比较，调用了线程池的addWorker方法，那么实际上也就是通过这里开启了线程池的worker来进行工作
来看看在worker的轮询中发生了什么：
1 final void runWorker(Worker w) {
2 Thread wt = Thread.currentThread();
3 Runnable task = w.firstTask;
4 w.firstTask = null;
5 w.unlock(); // allow interrupts
6 boolean completedAbruptly = true;
7 try {
8 while (task != null || (task = getTask()) != null) {
9 w.lock();
10 // If pool is stopping, ensure thread is interrupted;
11 // if not, ensure thread is not interrupted. This
12 // requires a recheck in second case to deal with
13 // shutdownNow race while clearing interrupt
14 if ((runStateAtLeast(ctl.get(), STOP) ||
15 (Thread.interrupted() &&
16 runStateAtLeast(ctl.get(), STOP))) &&
17 !wt.isInterrupted())
18 wt.interrupt();
19 try {
20 beforeExecute(wt, task);
21 Throwable thrown = null;
22 try {
23 task.run();
24 } catch (RuntimeException x) {
25 thrown = x; throw x;
26 } catch (Error x) {
27 thrown = x; throw x;
28 } catch (Throwable x) {
29 thrown = x; throw new Error(x);
30 } finally {
31 afterExecute(task, thrown);
32 }
33 } finally {
34 task = null;
35 w.completedTasks++;
36 w.unlock();
37 }
38 }
39 completedAbruptly = false;
40 } finally {
41 processWorkerExit(w, completedAbruptly);
42 }
43 }
可以看到在ThreadPoolExecutor的worker轮询线程中，会通过getTask方法，不断地从阻塞队列中获取任务
getTask方法：
1 private Runnable getTask() {
2 boolean timedOut = false; // Did the last poll() time out?
3
4 for (;;) {
5 int c = ctl.get();
6 int rs = runStateOf(c);
7
8 // Check if queue empty only if necessary.
9 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
10 decrementWorkerCount();
11 return null;
12 }
13
14 int wc = workerCountOf(c);
15
16 // Are workers subject to culling?
17 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
18
19 if ((wc > maximumPoolSize || (timed && timedOut))
20 && (wc > 1 || workQueue.isEmpty())) {
21 if (compareAndDecrementWorkerCount(c))
22 return null;
23 continue;
24 }
25
26 try {
27 Runnable r = timed ?
28 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
29 workQueue.take();
30 if (r != null)
31 return r;
32 timedOut = true;
33 } catch (InterruptedException retry) {
34 timedOut = false;
35 }
36 }
37 }
可以看到在这个方法中，在一系列的参数检查并设置完毕后，会通过workQueue的poll或者take方法来获取所需的任务
其中poll方法是在设置了超时时间的情况下进行获取，take则不带有超时时间
以take为例
DelayedWorkQueue的take方法：
1 public RunnableScheduledFuture<?> take() throws InterruptedException {
2 final ReentrantLock lock = this.lock;
3 lock.lockInterruptibly();
4 try {
5 for (;;) {
6 RunnableScheduledFuture<?> first = queue[0];
7 if (first == null)
8 available.await();
9 else {
10 long delay = first.getDelay(NANOSECONDS);
11 if (delay <= 0)
12 return finishPoll(first);
13 first = null; // don't retain ref while waiting
14 if (leader != null)
15 available.await();
16 else {
17 Thread thisThread = Thread.currentThread();
18 leader = thisThread;
19 try {
20 available.awaitNanos(delay);
21 } finally {
22 if (leader == thisThread)
23 leader = null;
24 }
25 }
26 }
27 }
28 } finally {
29 if (leader == null && queue[0] != null)
30 available.signal();
31 lock.unlock();
32 }
33 }
在for循环中首先取出数组中的第一个元素，也就是生成的小根堆中最小的那一个
得到first后，若是first为null，则说明当前没有可执行的任务，则使用available这个Condition对象，将AQS阻塞起来，等待下次任务创建时再通过前面提到的available唤醒阻塞
若是first存在，则通过getDelay方法获取时间间隔
getDelay方法：
1 public long getDelay(TimeUnit unit) {
2 return unit.convert(time - now(), NANOSECONDS);
3 }
这个方法就是用time减去当前时间now，得到的一个纳秒级时间差值
而time是在ScheduledFutureTask执行构造方法时，通过triggerTime方法，使用initialDelay进行计算出来的
triggerTime方法：
1 private long triggerTime(long delay, TimeUnit unit) {
2 return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
3 }
4
5 long triggerTime(long delay) {
6 return now() +
7 ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
8 }
9
10 private long overflowFree(long delay) {
11 Delayed head = (Delayed) super.getQueue().peek();
12 if (head != null) {
13 long headDelay = head.getDelay(NANOSECONDS);
14 if (headDelay < 0 && (delay - headDelay < 0))
15 delay = Long.MAX_VALUE + headDelay;
16 }
17 return delay;
18 }
可以看到time在这里实际上就是通过initialDelay加上当时设置的纳秒级时间组成的
其中overflowFree是为了防止Long类型的溢出做了一次计算，后边再说
所以take方法中，通过getDelay方法得到的是一个时间差，若是时间差小于等于0，则说明任务到了该执行的时候了，此时调用finishPoll
finishPoll方法：
1 private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
2 int s = --size;
3 RunnableScheduledFuture<?> x = queue[s];
4 queue[s] = null;
5 if (s != 0)
6 siftDown(0, x);
7 setIndex(f, -1);
8 return f;
9 }
这个方法的逻辑还是比较简单的，就是一个简单的小根堆重新调整的操作，由于f需要被取出，此时利用最后一个元素，完成一次自上向下的调整（生成时是自下向上）
siftDown方法和siftUp类似：
1 private void siftDown(int k, RunnableScheduledFuture<?> key) {
2 int half = size >>> 1;
3 while (k < half) {
4 int child = (k << 1) + 1;
5 RunnableScheduledFuture<?> c = queue[child];
6 int right = child + 1;
7 if (right < size && c.compareTo(queue[right]) > 0)
8 c = queue[child = right];
9 if (key.compareTo(c) <= 0)
10 break;
11 queue[k] = c;
12 setIndex(c, k);
13 k = child;
14 }
15 queue[k] = key;
16 setIndex(key, k);
17 }
由二叉树性质half 保证只操作到倒数第二层
在循环中，首先根据k（当前也就是根节点），得到其左右孩子的下标
若是右孩子存在，那么就用左孩子和右孩子比较，选出最下的哪一个作为child
若是右孩子不存在，则直接使用左孩子作为child
当选出child后，再和待插入的元素key比较
若是key小，则结束循环，直接将key插入k所在位置
若不是，则将当前child所在元素放在k所在位置，然后从child位置继续开始向下寻找，直到找到一个大于key或者遍历完毕
这样自上向下的将当前堆又调整成了小根堆，以后的定时周期任务都是以这种方式来调用的
看到这ScheduledThreadPoolExecutor的定时周期任务已经基本理解了，只不过还存在一个问题，当执行周期任务，会从小根堆取出，那么该任务下一次的执行时间何时更新到小根堆？
回到ThreadPoolExecutor的worker的runWorker方法中，在调用完getTask方法后，在进行完一系列完全检查后，会直接调用task的run方法，而此时的task是经过之前ScheduledFutureTask包装的
ScheduledFutureTask的run方法：
1 public void run() {
2 boolean periodic = isPeriodic();
3 if (!canRunInCurrentRunState(periodic))
4 cancel(false);
5 else if (!periodic)
6 ScheduledFutureTask.super.run();
7 else if (ScheduledFutureTask.super.runAndReset()) {
8 setNextRunTime();
9 reExecutePeriodic(outerTask);
10 }
11 }
若是设置了周期任务（period不为0），那么isPeriodic方法为true
逻辑上就会执行runAndReset方法，这个方法内部就会调用我们传入的Runnable的run方法，从而真正地执行我们的任务
在执行完毕后，可以看到调用了setNextRunTime方法
setNextRunTime方法：
1 private void setNextRunTime() {
2 long p = period;
3 if (p > 0)
4 time += p;
5 else
6 time = triggerTime(-p);
7 }
这里就很简单，利用当前time和period计算出下一次的time
由于scheduleWithFixedDelay和scheduleAtFixedRate之前所说的不一样之处，在这里就得到了体现
因为scheduleAtFixedRate的period是大于0的，所以scheduleAtFixedRate计算出来的时间间隔就是initialDelay + n*period的这种形式，那么其执行就会有固定的时间点，不过这还是要取决于任务的执行时间，若是任务的执行时间大于时间间隔，那么当上一次任务执行完毕，就会立刻执行，而不是等到时间点到了，若是任务的执行时间小于时间间隔，那么毫无疑问就需要等到时间点到了才执行下一次的任务
由于scheduleWithFixedDelay的period是小于0的，所以需要执行triggerTime
triggerTime方法：
1 long triggerTime(long delay) {
2 return now() +
3 ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
4 }
可以看到若是不存在Long类型的溢出问题，那么下一次的时间就等于当前时间加时间间隔，所以说scheduleWithFixedDelay的不同之处在于其算上了任务的实际执行时间
若是存在Long类型的溢出问题时
在overflowFree中：
1 private long overflowFree(long delay) {
2 Delayed head = (Delayed) super.getQueue().peek();
3 if (head != null) {
4 long headDelay = head.getDelay(NANOSECONDS);
5 if (headDelay < 0 && (delay - headDelay < 0))
6 delay = Long.MAX_VALUE + headDelay;
7 }
8 return delay;
9 }
首先通过peek得到队列中的第一个元素，若是不存在，则直接返回delay
若是存在，通过getDelay得到headDelay
这里就会存在两情况
任务还没达到执行时间，则headDelay 大于零
任务达到执行时间，但却由于之前的任务还没执行完毕，遭到了延时，headDelay 小于0
所以这次的计算就是将headDelay这部分超时时间减去，以防止后续影响compareTo的比较，从而引起offer顺序的错误
（只不过这种情况正常不会遇见。。。）
在计算完成下一次的运行时间后
调用reExecutePeriodic方法：
1 void reExecutePeriodic(RunnableScheduledFuture<?> task) {
2 if (canRunInCurrentRunState(true)) {
3 super.getQueue().add(task);
4 if (!canRunInCurrentRunState(true) && remove(task))
5 task.cancel(false);
6 else
7 ensurePrestart();
8 }
9 }
其中传入的这个task（outerTask）其实就是当前执行完毕的这个任务，
可以看到这里canRunInCurrentRunState成立的情况下，就会通过
getQueue得到阻塞队列，再次通过DelayedWorkQueue的add方法将其加入到小根堆中，只不过这时的time发生了变化
若是情况正常，则继续通过ThreadPoolExecutor的ensurePrestart方法，调度worker的工作
这样定时周期任务就能正常执行
ScheduledThreadPoolExecutor分析到此结束
加载全部内容