ScheduledThreadPoolExecutor详解
Posted zhangxufeng
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ScheduledThreadPoolExecutor详解相关的知识，希望对你有一定的参考价值。
本文主要分为两个部分，第一部分首先会对ScheduledThreadPoolExecutor进行简单的介绍，并且会介绍其主要API的使用方式，然后介绍了其使用时的注意点，第二部分则主要对ScheduledThreadPoolExecutor的实现细节进行介绍。
1. 使用简介
ScheduledThreadPoolExecutor是一个使用线程池执行定时任务的类，相较于Java中提供的另一个执行定时任务的类Timer，其主要有如下两个优点：
- 使用多线程执行任务，不用担心任务执行时间过长而导致任务相互阻塞的情况，Timer是单线程执行的，因而会出现这个问题；
- 不用担心任务执行过程中，如果线程失活，其会新建线程执行任务，Timer类的单线程挂掉之后是不会重新创建线程执行后续任务的。
除去上述两个优点外，ScheduledThreadPoolExecutor还提供了非常灵活的API，用于执行任务。其任务的执行策略主要分为两大类：①在一定延迟之后只执行一次某个任务；②在一定延迟之后周期性的执行某个任务。如下是其主要API：
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay, long delay, TimeUnit unit);
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay, long period, TimeUnit unit);上述四个方法中，第一个和第二个方法属于第一类，即在delay指定的延迟之后执行第一个参数所指定的任务，区别在于，第二个方法执行之后会有返回值，而第一个方法执行之后是没有返回值的。第三个和第四个方法则属于第二类，即在第二个参数（initialDelay）指定的时间之后开始周期性的执行任务，执行周期间隔为第三个参数指定的时间，但是这两个方法的区别在于第三个方法执行任务的间隔是固定的，无论上一个任务是否执行完成，而第四个方法的执行时间间隔是不固定的，其会在周期任务的上一个任务执行完成之后才开始计时，并在指定时间间隔之后才开始执行任务。如下是使用scheduleWithFixedDelay()和scheduleAtFixedRate()方法编写的测试用例：
public class ScheduledThreadPoolExecutorTest {
private ScheduledThreadPoolExecutor executor;
private Runnable task;
@Before
public void before() {
executor = initExecutor();
task = initTask();
}
private ScheduledThreadPoolExecutor initExecutor() {
return new ScheduledThreadPoolExecutor(2);;
}
private Runnable initTask() {
long start = System.currentTimeMillis();
return () -> {
print("start task: " + getPeriod(start, System.currentTimeMillis()));
sleep(SECONDS, 10);
print("end task: " + getPeriod(start, System.currentTimeMillis()));
};
}
@Test
public void testFixedTask() {
print("start main thread");
executor.scheduleAtFixedRate(task, 15, 30, SECONDS);
sleep(SECONDS, 120);
print("end main thread");
}
@Test
public void testDelayedTask() {
print("start main thread");
executor.scheduleWithFixedDelay(task, 15, 30, SECONDS);
sleep(SECONDS, 120);
print("end main thread");
}
private void sleep(TimeUnit unit, long time) {
try {
unit.sleep(time);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private int getPeriod(long start, long end) {
return (int)(end - start) / 1000;
}
private void print(String msg) {
System.out.println(msg);
}
}可以看到，上述两个测试用例代码块基本是一致的，区别在于第一个用例调用的是scheduleAtFixedRate()方法，而第二个用例调用的是scheduleWithFixedDelay()。这里两个用例都是设置的在延迟15s后每个30s执行一次指定的任务，而该任务执行时长为10s。如下分别是这两个测试用例的执行结果：
start main thread
start task: 15
end task: 25
start task: 45
end task: 55
start task: 75
end task: 85
start task: 105
end task: 115
end main threadstart main thread
start task: 15
end task: 25
start task: 55
end task: 65
start task: 95
end task: 105
end main thread对比上述执行结果可以看出，对于scheduleAtFixedRate()方法，其每次执行任务的开始时间间隔都为固定不变的30s，与任务执行时长无关，而对于scheduleWithFixedDelay()方法，其每次执行任务的开始时间间隔都为上次任务执行时间加上指定的时间间隔。
这里关于ScheduledThreadPoolExecutor的使用有三点需要说明如下：
- ScheduledThreadPoolExecutor继承自ThreadPoolExecutor（ThreadPoolExecutor详解），因而也有继承而来的execute()和submit()方法，但是ScheduledThreadPoolExecutor重写了这两个方法，重写的方式是直接创建两个立即执行并且只执行一次的任务；
- ScheduledThreadPoolExecutor使用ScheduledFutureTask封装每个需要执行的任务，而任务都是放入DelayedWorkQueue队列中的，该队列是一个使用数组实现的优先队列，在调用ScheduledFutureTask::cancel()方法时，其会根据removeOnCancel变量的设置来确认是否需要将当前任务真正的从队列中移除，而不只是标识其为已删除状态；
- ScheduledThreadPoolExecutor提供了一个钩子方法decorateTask(Runnable, RunnableScheduledFuture)用于对执行的任务进行装饰，该方法第一个参数是调用方传入的任务实例，第二个参数则是使用ScheduledFutureTask对用户传入任务实例进行封装之后的实例。这里需要注意的是，在ScheduledFutureTask对象中有一个heapIndex变量，该变量用于记录当前实例处于队列数组中的下标位置，该变量可以将诸如contains()，remove()等方法的时间复杂度从O(N)降低到O(logN)，因而效率提升是比较高的，但是如果这里用户重写decorateTask()方法封装了队列中的任务实例，那么heapIndex的优化就不存在了，因而这里强烈建议是尽量不要重写该方法，或者重写时也还是复用ScheduledFutureTask类。
2. 源码详解
2.1 主要属性
ScheduledThreadPoolExecutor主要有四个属性，分别如下：
private volatile boolean continueExistingPeriodicTasksAfterShutdown;
private volatile boolean executeExistingDelayedTasksAfterShutdown = true;
private volatile boolean removeOnCancel = false;
private static final AtomicLong sequencer = new AtomicLong();- continueExistingPeriodicTasksAfterShutdown：用于标识当前Executor对象shutdown时，是否继续执行已经存在于任务队列中的定时任务（调用scheduleAtFixedRate()方法生成的任务）；
- executeExistingDelayedTasksAfterShutdown：用于标识当前Executor对象shutdown时，是否继续执行已经存在于任务队列中的定时任务（调用scheduleWithFixedDelay()方法生成的任务）；
- removeOnCancel：用于标识如果当前任务已经取消了，是否将其从任务队列中真正的移除，而不只是标识其为删除状态；
- sequencer：其为一个AtomicLong类型的变量，该变量记录了当前任务被创建时是第几个任务的一个序号，这个序号的主要用于确认当两个任务开始执行时间相同时具体哪个任务先执行，比如两个任务的开始执行时间都为1515847881158，那么序号小的任务将先执行。
2.2 ScheduledFutureTask
在ScheduledThreadPoolExecutor中，主要使用ScheduledFutureTask封装需要执行的任务，该类的主要声明如下：
private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {
private final long sequenceNumber; // 记录当前实例的序列号
private long time; // 记录当前任务下次开始执行的时间
// 记录当前任务执行时间间隔，等于0则表示当前任务只执行一次，大于0表示当前任务为fixedRate类型的任务，
// 小于0则表示其为fixedDelay类型的任务
private final long period;
RunnableScheduledFuture<V> outerTask = this; // 记录需要周期性执行的任务的实例
int heapIndex; // 记录当前任务在队列数组中位置的下标
ScheduledFutureTask(Runnable r, V result, long ns, long period) {
super(r, result);
this.time = ns;
this.period = period;
this.sequenceNumber = sequencer.getAndIncrement(); // 序号在创建任务实例时指定，且后续不会变化
}
public long getDelay(TimeUnit unit) {
return unit.convert(time - now(), NANOSECONDS);
}
// 各个任务在队列中的存储方式是一个基于时间和序号进行比较的优先队列，当前方法定义了优先队列中两个
// 任务执行的先后顺序。这里先对两个任务开始执行时间进行比较，时间较小者优先执行，若开始时间相同，
// 则比较两个任务的序号，序号小的任务先执行
public int compareTo(Delayed other) {
if (other == this)
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;
}
public boolean isPeriodic() { // 判断是否为周期性任务
return period != 0;
}
// 当前任务执行之后，会判断当前任务是否为周期性任务，如果为周期性任务，那么就调用当前方法计算
// 当前任务下次开始执行的时间。这里如果当前任务是fixedRate类型的任务(p > 0)，那么下次执行时间
// 就是此次执行的开始时间加上时间间隔，如果当前任务是fixedDelay类型的任务(p < 0)，那么下次执行
// 时间就是当前时间(triggerTime()方法会获取系统当前时间)加上任务执行时间间隔。可以看到，定频率
// 和定延迟的任务的执行时间区别就在当前方法中进行了指定，因为调用当前方法时任务已经执行完成了，
// 因而triggerTime()方法中获取的时间就是任务执行完成之后的时间点
private void setNextRunTime() {
long p = period;
if (p > 0)
time += p;
else
time = triggerTime(-p);
}
// 取消当前任务的执行，super.cancel(boolean)方法也即FutureTask.cancel(boolean)方法。该方法传入
// true表示如果当前任务正在执行，那么立即终止其执行；传入false表示如果当前方法正在执行，那么等待其
// 执行完成之后再取消当前任务。
public boolean cancel(boolean mayInterruptIfRunning) {
boolean cancelled = super.cancel(mayInterruptIfRunning);
// 判断是否设置了取消后移除队列中当前任务，是则移除当前任务
if (cancelled && removeOnCancel && heapIndex >= 0)
remove(this);
return cancelled;
}
public void run() {
boolean periodic = isPeriodic(); // 判断是否为周期性任务
if (!canRunInCurrentRunState(periodic)) // 判断是否能够在当前状态下执行该任务
cancel(false);
else if (!periodic) // 如果能执行当前任务，但是任务不是周期性的，那么就立即执行该任务一次
ScheduledFutureTask.super.run();
else if (ScheduledFutureTask.super.runAndReset()) { // 是周期性任务，则立即执行当前任务并且重置
setNextRunTime(); // 在当前任务执行完成后调用该方法计算当前任务下次执行的时间
reExecutePeriodic(outerTask); // 将当前任务放入任务队列中以便下次执行
}
}
}在ScheduledFutureTask中，主要有三个点需要强调：
- 对于run()方法的第一个分支，canRunInCurrentRunState()方法的声明如下所示，可以看到，该方法是用于判断当前任务如果为周期性任务，那么其是否允许在shutdown状态下继续执行已经存在的周期性任务，是则表示当前状态下是可以执行当前任务的，这里isRunningOrShutdown()方法继承自ThreadPoolExecutor;
boolean canRunInCurrentRunState(boolean periodic) {
return isRunningOrShutdown(periodic ?
continueExistingPeriodicTasksAfterShutdown :
executeExistingDelayedTasksAfterShutdown);
}- 在run()方法的最后一个if分支中，其首先会执行当前任务，在执行完成时才会调用setNextRunTime()方法设置下次任务执行时间，也就是说对于fixedRate和fixedDelay类型的任务都是在这个时间点才设置的，因而虽然fixedRate类型的任务，即使该任务下次执行时间比当前时间要早，其也只会在当前任务执行完成后立即执行，而不会与当前任务还未执行完时就执行；对于fixedDelay任务则不会存在该问题，因为其是以任务完成后的时间点为基础计算下次执行的时间点；
- 对于run()方法的最后一个分支中的reExecutePeriodic()方法，其会将当前任务加入到任务队列中，并且调用父类的ensurePrestart()方法确保有可用的线程来执行当前任务，如下是该方法的具体实现：
void reExecutePeriodic(RunnableScheduledFuture<?> task) {
if (canRunInCurrentRunState(true)) { // 判断当前任务是否可以继续执行
super.getQueue().add(task); // 将当前任务加入到任务队列中
if (!canRunInCurrentRunState(true) && remove(task)) // 双检查法判断任务在加入过程中是否取消了
task.cancel(false);
else
ensurePrestart(); // 初始化核心线程等确保任务可以被执行
}
}从ScheduledFutureTask的实现总结来看，当每创建一个该类实例时，会初始化该类的一些主要属性，如下次开始执行的时间和执行的周期。当某个线程调用该任务，即执行该任务的run()方法时，如果该任务不为周期性任务，那么执行该任务之后就不会有其余的动作，如果该任务为周期性任务，那么在将当前任务执行完毕之后，还会重置当前任务的状态，并且计算下次执行当前任务的时间，然后将其放入队列中以便下次执行。
2.3 DelayedWorkQueue
DelayedWorkQueue的实现与DelayQueue以及PriorityQueue的实现基本相似，形式都为一个优先队列，并且底层是使用堆结构来实现优先队列的功能，在数据存储方式上，其使用的是数组来实现。这里DelayedWorkQueue与DelayQueue以及PriorityQueue不同的点在于DelayedWorkQueue中主要存储ScheduledFutureTask类型的任务，该任务中有一个heapIndex属性保存了当前任务在当前队列数组中的位置下标，其主要提升的是对队列的诸如contains()和remove()等需要定位当前任务位置的方法的效率，时间复杂度可以从O(N)提升到O(logN)。如下是DelayedWorkQueue的实现代码（这里只列出了该类的主要属性和与实现ScheduledThreadPoolExecutor功能相关的方法，关于如何使用数组实现优先队列请读者查阅相关文档）：
static class DelayedWorkQueue extends AbstractQueue<Runnable> implements BlockingQueue<Runnable> {
private static final int INITIAL_CAPACITY = 16; // 数组初始化大小
private RunnableScheduledFuture<?>[] queue = new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
private final ReentrantLock lock = new ReentrantLock(); // 对添加和删除元素所使用的锁
private int size = 0; // 当前队列中有效任务的个数
private Thread leader = null; // 执行队列头部任务的线程
private final Condition available = lock.newCondition(); // 除leader线程外其余线程的等待队列
// 在对任务进行移动时，判断其是否为ScheduledFutureTask实例，如果是则维护其heapIndex属性
private void setIndex(RunnableScheduledFuture<?> f, int idx) {
if (f instanceof ScheduledFutureTask)
((ScheduledFutureTask)f).heapIndex = idx;
}
private void siftUp(int k, RunnableScheduledFuture<?> key) {/* 省略 */}
private void siftDown(int k, RunnableScheduledFuture<?> key) {/* 省略 */}
private int indexOf(Object x) {
if (x != null) {
if (x instanceof ScheduledFutureTask) { // 如果为ScheduledFutureTask则可返回其heapIndex属性
int i = ((ScheduledFutureTask) x).heapIndex;
if (i >= 0 && i < size && queue[i] == x)
return i;
} else { // 如果不为ScheduledFutureTask实例，则需要遍历队列查询当前元素的位置
for (int i = 0; i < size; i++)
if (x.equals(queue[i]))
return i;
}
}
return -1;
}
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;
if (i >= queue.length)
grow(); // 队列容量不足，对其进行扩容
size = i + 1;
if (i == 0) { // 如果其为队列第一个元素，则将其放入队列头部
queue[0] = e;
setIndex(e, 0);
} else { //如果不为第一个元素，则通过堆的上移元素操作移动当前元素至合适的位置
siftUp(i, e);
}
if (queue[0] == e) { // 如果被更新的是队列头部元素，则更新记录的执行头部任务的线程
leader = null;
available.signal();
}
} finally {
lock.unlock();
}
return true;
}
// 完成从队列拉取元素操作，并且将其从队列中移除
private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
int s = --size;
RunnableScheduledFuture<?> x = queue[s];
queue[s] = null; // 将队列最尾部的元素置空
if (s != 0) // 将最后一个元素放入第一个位置，并且将其下推至合适的位置
siftDown(0, x); // 这里idx置为0是因为当前方法的入参f都为队列的第一个元素
setIndex(f, -1);
return f;
}
// 尝试从队列(堆)中获取元素，如果没有元素或者元素的延迟时间还未到则返回空
public RunnableScheduledFuture<?> poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
RunnableScheduledFuture<?> first = queue[0];
// 在此处代码控制了当从堆顶拉取元素时，如果元素的延迟时间还未达到，则不返回当前元素
if (first == null || first.getDelay(NANOSECONDS) > 0)
return null;
else
return finishPoll(first); // 返回堆顶元素
} finally {
lock.unlock();
}
}
// 通过无限for循环获取堆顶的元素，这里take()方法会阻塞当前线程，直至获取到了可执行的任务。
// 可以看到，在第一次for循环中，如果堆顶不存在任务，则其会加入阻塞队列中，如果存在任务，但是
// 其延迟时间还未到，那么当前线程会等待该延迟时间长的时间，然后查看任务是否可用，当获取到任务
// 之后，其会将其从队列中移除，并且唤醒等待队列中其余等待的线程执行下一个任务
public RunnableScheduledFuture<?> take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
RunnableScheduledFuture<?> first = queue[0];
if (first == null)
available.await(); // 堆内没有元素，当前线程进入等待队列中
else {
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0) // 堆顶元素延迟时间小于0，可立即获取任务
return finishPoll(first);
first = null;
if (leader != null)
available.await(); // 已经有线程在等待堆顶元素，则当前线程进入等待队列中
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay); // 当前线程等待一定时长后获取任务并执行
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && queue[0] != null)
available.signal(); // 当前线程获取完任务之后唤醒等待队列中的下一个线程执行下一个任务
lock.unlock();
}
}
}从DelayedWorkQueue的take()和poll()方法可以看出来，对于队列中任务的等待时间的限制主要是在这两个方法中实现的，如果任务的等待时间还未到，那么该方法就会阻塞线程池中的线程，直至任务可以执行。
2.4 scheduleAtFixedRate()和scheduleWithFixedDelay()方法
前面我们对ScheduledThreadPoolExecutor的主要属性和主要内部类都进行了详细的讲解，基本上已经可以看出其是如何实现定时执行任务的功能的，接下来我们主要对客户端可以调用的主要方法进行简要介绍，这里scheduleAtFixedRate()和scheduleWithFixedDelay()方法的实现基本是一致的，两个方法最细微的区别在于ScheduledFutureTask的setNextRunTime()方法的实现，该方法的实现前面已经进行了讲解，我们这里则以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; // 初始化周期任务属性outerTask
delayedExecute(t); // 执行该任务
return t;
}从上述代码可以看出来，scheduleAtFixedRate()首先对客户端任务实例进行了封装，装饰，并且初始化了封装后的任务实例的outerTask属性，最后调用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
ensurePrestart(); // 若线程池没有初始化，则进行一些初始化工作
}
}上述方法为主要的执行任务的方法，该方法首先会将任务加入到任务队列中，如果线程池已经初始化过，那么该任务就会有等待的线程执行该任务。在加入到任务队列之后通过双检查法检查线程池是否已经shutdown了，如果是则将该任务从任务队列中移除。如果当前线程池没有shutdown，就调用继承自ThreadPoolExecutor的ensurePrestart()方法，该方法会对线程池进行一些初始化工作，如初始化核心线程，然后各个线程会调用上述等待队列的take()方法获取任务执行。
以上是关于ScheduledThreadPoolExecutor详解的主要内容，如果未能解决你的问题，请参考以下文章