ScheduledThreadPoolExecutor原理和源码
延迟以及周期性执行线程池。
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
堵塞队列为DelayedWorkQueue,这是ScheduledThreadPoolExecutor的内部类。看其grow扩容方法会发现,这个队列最大大小为Integer.MAX_VALUE。所以队列可以放置很多元素。
看堵塞队列的DelayedWorkQueue的offer方法
可以看出offer的元素是RunnableScheduledFuture类型。详细的流程不看,设计堆的调整等操作,目的是保证一个有优先级级的队列,延迟时间短的优先级高,在队列的最前面。
public boolean offer(Runnable x) {
if (x == null)
throw new NullPointerException();
// offer的元素是RunnableScheduledFuture类型
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;
}
看DelayedWorkQueue队列保存的元素类型RunnableScheduledFuture。
RunnableScheduledFuture的唯一一个自己的方法是isPeriodic,判断这个任务时候是周期性任务。
RunnableScheduledFuture的父类是RunnableFuture继承了Runnable接口,所以RunnableScheduledFuture也可以看成一个线程。
还有一个父类是ScheduledFuture,其继承了Delayed接口,Delayed接口唯一的方法就是获取任务的剩余延迟时间,以供延迟队列的延迟获取元素。
所以RunnableScheduledFuture类型元素,即是一个线程,有run方法,也可以判断是否周期性执行,又可以获取任务剩余延迟时间。
public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {
/**
* Returns {@code true} if this task is periodic. A periodic task may
* re-run according to some schedule. A non-periodic task can be
* run only once.
*
* @return {@code true} if this task is periodic
*/
boolean isPeriodic();
}
public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}
public interface ScheduledFuture<V> extends Delayed, Future<V> {
}
看堵塞队列的take方法
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);
// 剩余延迟时间小于0,则finishPoll调整堆,然后直接返回这个元素
if (delay <= 0)
return finishPoll(first);
first = null; // don't retain ref while waiting
// 剩余延迟时间大于0。如果leader不为空,说明已经有线程成为leader并等待堆顶任务
if (leader != null)
available.await();
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// 堵塞等待delay时间
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && queue[0] != null)
available.signal();
lock.unlock();
}
}
take方法总结:take方法和普通的延迟队列一样,比如delayQueue。等待队列头部元素,剩余延迟时间过了之后,才能获取到值。
DelayedWorkQueue延迟队列分析完了之后,分析ScheduledThreadPoolExecutor的关键方法。
构造方法:除了堵塞队列以外,和ThreadPoolExcutor差不多,但是由于DelayedWorkQueue堵塞队列是无限大的,所以,不存在最大线程数以及最大线程数空闲时间。
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
execute方法:延迟时间为0,所以正常执行,该启动新线程则启动新线程,该放入队列则放入队列。
public void execute(Runnable command) {
schedule(command, 0, NANOSECONDS);
}
public ScheduledFuture<?> schedule(Runnable command,
long delay,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
// decorateTask方法只是返回最后一个入参,什么都没做。
// ScheduledFutureTask类是RunnableScheduledFuture的实现类,所以ScheduledFutureTask类可以放入DelayWorkQueue堵塞队列
RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Void>(command, null,
triggerTime(delay, unit)));
// delayedExecute方法把RunnableScheduledFuture放入堵塞队列,如果线程数小于核心线程,则开启线程从队列中取任务
delayedExecute(t);
return t;
}
// ScheduledFutureTask构造方法,period为0,说明不是周期性任务,time表示延迟时间。
ScheduledFutureTask(Runnable r, V result, long ns) {
super(r, result);
this.time = ns;
this.period = 0;
this.sequenceNumber = sequencer.getAndIncrement();
}
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();
}
}
void ensurePrestart() {
int wc = workerCountOf(ctl.get());
// 已经开启线程数小于核心线程数,则开启新线程,
// addWorker传入的第一参数为null,也就是worker的firstTask为null,所以直接从getTask()直接从队列中获取任务,
// 堵塞队列是延迟队列,所以可能会延迟获取到任务。
if (wc < corePoolSize)
addWorker(null, true);
else if (wc == 0)
addWorker(null, false);
}
之后的流程就是,和和普通线程池一样,ThreadPoolExecutor的真正工作的内部类worker,其就是开启的线程类的runnable入参,线程start启动的时候,会运行workker的run方法,workker的run方法会循环从队列中获取任务执行,因为是延迟队列,所以会又延迟的效果,延迟的原理前面分析堵塞队列的时候分析了,condition.awaitNanos方法实现延迟等待。
获取任务后,运行任务的run方法。
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);
}
}
scheduleAtFixedRate:该方法在initialDelay时长后第一次执行任务,以后每隔period时长,再次执行任务。注意,period是从任务开始执行算起的。开始执行任务后,定时器每隔period时长检查该任务是否完成,如果完成则再次启动任务,否则等该任务结束后才再次启动任务,看下图示例。
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(period)
scheduleWithFixDelay:该方法在initialDelay时长后第一次执行任务,以后每当任务执行完成后,等待delay时长,再次执行任务,看下图示例。
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(-delay)
和scheduledAtFixedRate类似,唯一不同的地方在于在于创建的ScheduledFutureTask不同