Java并发编程—ScheduledThreadPoolExecutor原理分析

原文作者：小付　

原文地址：ScheduledThreadPoolExecutor原理分析

一、简单使用

二、类UML图

三、处理流程

四、任务提交方式

五、SchduledFutureTask之run方法实现

一、简单使用

这里先学会简单使用再深入探讨。

        ScheduledThreadPoolExecutor  scheduled = new ScheduledThreadPoolExecutor(2);scheduled.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {loge("time:");}}, 0, 40, TimeUnit.MILLISECONDS);//0表示首次执行任务的延迟时间，40表示每次执行任务的间隔时间，TimeUnit.MILLISECONDS执行的时间间隔数值单位

间隔单位毫秒：TimeUnit.MILLISECONDS
间隔单位秒：TimeUnit.SECONDS
间隔单位分钟：TimeUnit.MINUTES
间隔单位小时：TimeUnit.HOURS
间隔单位天：TimeUnit.DAYS

二、类UML图

定时线程池类ScheduledThreadPoolExecutor用来处理延时任务或定时任务，它的类结构图如下：

三、处理流程

它的处理流程如下：主线程提交的定时任务或延时任务采用DelayQueue存储。
DelayQueue内部封装了一个PriorityQueue（优先队列，是一种堆结构），它会根据time的先后时间排序，若 time相同则根据sequenceNumber排序；DelayQueue也是一个无界队列；
工作线程会从DelayQueue取已经到期的任务去执行；
执行结束后重新设置任务的到期时间，再次放回DelayQueue

PriorityQueue堆结构如下图：

可见，DelayedQueue是一个基于最小堆结构的队列。堆结构可以使用数组表示，可以转换成如下的数组：

为什么要使用DelayedWorkQueue呢？定时任务执行时需要取出最近要执行的任务，所以任务在队列中每次出队时一定要是当前队列中执行时间最靠前的，所以自然要使用优先级队列。DelayedQueue是一个优先级队列，它可以保证每次出队的任务都是当前队列中执行时间最靠前的，由于它是基于堆结构的队列，堆结构在执行插入和删除操作时的最坏时间复杂度是 O(logN)。PriorityQueue内部的比较逻辑实现代码如下：

    public int compareTo(Delayed other) {if (other == this) // compare zero if same objectreturn 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;elsereturn 1;}long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;}

四、任务提交方式

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

1、schedule()方法

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

public ScheduledFuture<?> schedule(Runnable command,long delay,TimeUnit unit) {//参数校验if (command == null || unit == null)throw new NullPointerException();//这里是一个嵌套结构，首先把用户提交的任务包装成ScheduledFutureTask 8 //然后在调用decorateTask进行包装，该方法是留给用户去扩展的，默认是个 空方法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 {//与ThreadPoolExecutor不同，这里直接把任务加入延迟队列super.getQueue().add(task);//如果当前状态无法执行任务，则取消if (isShutdown() &&!canRunInCurrentRunState(task.isPeriodic()) &&remove(task))task.cancel(false);else//这里是增加一个worker线程，避免提交的任务没有worker去执行//原因就是该类没有像ThreadPoolExecutor一样，woker满了才放入队列ensurePrestart();}}

2、scheduleAtFixedRate()

schedule和scheduleAtFixedRate的区别在于，如果指定开始执行的时间在当前系统运行时间之前，scheduleAtFixedRate会把已经过去的时间也作为周期执行，而schedule不会把过去的时间算上。scheduleAtFixedRate()源码如下：

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

SchduledFutureTask：从上面提交任务方法的源码的第六行ScheduledFutureTask<Void> sft =new ScheduledFutureTask<Void>(command,null,triggerTime(initialDelay, unit), unit.toNanos(period));可以看到scheduleAtFixedRate()方法将command封装为ScheduledFutureTask。下面再来看一下SchduledFutureTask接收的参数(成员变量)：

 - private long time：任务开始的时间 - private final long sequenceNumber;：任务的序号 - private final long period：任务执行的时间间隔

五、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；
如果不是周期性任务，调用FutureTask中的run方法执行，会设置执行结果，然后直接返回，否则执行步骤3；
如果是周期性任务，调用FutureTask中的runAndReset方法执行，不会设置执行结果，然后直接返回，否则执行步骤4和步骤5；
计算下次执行该任务的具体时间；
重复执行任务。

reExecutePeriodic方法

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

该方法和delayedExecute方法类似，不同的是：

由于调用reExecutePeriodic方法时已经执行过一次周期性任务了，所以不会 reject当前任务；
传入的任务一定是周期性任务。

一切伟大的行动和思想，都有一个微不足道的开始。