ScheduledThreadPoolExecutor作为ScheduledExecutorService接口的实现，提供了延迟执行任务或者周期性执行任务的能力。通过名称可以看出，ScheduledThreadPoolExecutor基于线程池实现，它通过继承ThreadPoolExecutor实现线程池管理能力的复用，同时扩展了自己的定时任务调度能力。
首先来看ScheduledExecutorServicej接口，它继承了ExecutorService接口，作为任务执行器的一种扩展类型，提供了如下方法：
ScheduledThreadPoolExecutor继承ThreadPoolExecutor，主要做了如下改变：
ScheduledThreadPoolExecutor由3部分组成：
ScheduledThreadPoolExecutor的构造器最多指定3个参数：
ScheduledThreadPoolExecutor构造器会调用父类构造器进行线程池初始化，使用DelayedWorkQueue作为阻塞队列，该队列为无界队列，因而maximumPoolSize属性配置无效。又因为都是核心工作线程，没有非核心线程需要回收，因而keepAliveTime配置为0。代码如下：
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue(), threadFactory, handler);
}ScheduledThreadPoolExecutor初始化时并不会预先创建工作线程，而是在提交任务的时候，通过父类java.util.concurrent.ThreadPoolExecutor#ensurePrestart方法判断线程数是否达到corePoolSize，如果未达到，则新增线程；实现逻辑如下：
void ensurePrestart() { int wc = workerCountOf(ctl.get()); // 当前线程数 < corePoolSize，则添加核心工作线程； if (wc < corePoolSize) addWorker(null, true); // 如果0 == wc >= corePoolSize，则表示corePoolSize配置为0，则初始化一个非核心工作线程； else if (wc == 0) addWorker(null, false); }
ScheduledThreadPoolExecutor的任务执行分为单次执行和周期性执行。
public void execute(Runnable command) {
schedule(command, 0, NANOSECONDS);
}无论是单次执行还是周期性执行，其执行的入口都是delayedExecute方法。delayedExecute()将任务放入到阻塞队列中，复用ThreadPoolExecutor的逻辑进行任务调度。代码如下：
private void delayedExecute(RunnableScheduledFuture<?> task) {
// 如果调度器关闭，则拒绝接收任务
if (isShutdown())
// 执行拒绝策略
reject(task);
// 如果调度器未关闭
else {
// 添加任务到阻塞队列，等待调度执行；
super.getQueue().add(task);
// 如果此时调度器关闭，则取消任务；
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
// 如果调度器正常运行，则检查开启线程数是否达到corePoolSize
// 如果未达到corePoolSize，则初始化1个工作线程；
// 如果corePoolSize设置为0，则会初始化1个非core工作线程；
else
ensurePrestart();
}
}当ThreadPoolExecutor的Worker线程从阻塞队列取出任务执行时，会调用ScheduledFutureTask的run方法。该方法对任务类型进行判断，如果是单次执行任务，则立即执行并设置返回结果。如果是周期性执行任务，则执行任务并设置下一次执行时间，然后将任务放入到阻塞队列中，等待下一次调度。方法代码如下：
public void run() {
boolean periodic = isPeriodic();
// 如果当前不可运行任务，则取消任务
if (!canRunInCurrentRunState(periodic))
cancel(false);
// 如果是单次执行的任务
else if (!periodic)
// 直接调用FutureTask.run()方法执行；
ScheduledFutureTask.super.run();
// 如果是周期性任务，则运行任务但不设置结果；
else if (ScheduledFutureTask.super.runAndReset()) {
// 设置任务下一次执行时间
setNextRunTime();
// 将任务重新加入队列，等待下一次调度
reExecutePeriodic(outerTask);
}
}schedule的执行主要分为参数封装和执行两个步骤。实现如下：
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;
}参数封装过程会调用decorateTask方法，该方法为protected的空方法，用于定制RunnableScheduledFuture的属性，可以通过重写实现定制。
protected <V> RunnableScheduledFuture<V> decorateTask(
Runnable runnable, RunnableScheduledFuture<V> task) {
return task;
}scheduleAtFixedRate()的实现与schedule()方法非常相似，仅是将decorateTask()返回的RunnableScheduledFuture对象设置为原有Future的outerTask属性。在重新知心任务时，会将outerTask添加到阻塞队列，从而保证decorateTask()的定制效果一直有效。
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);
// 保存定制后的Future对象，便于再次调用；
sft.outerTask = t;
delayedExecute(t);
return t;
}scheduleAtFixedRate()的实现与schedule()方法非常相似，仅是设置ScheduledFutureTask延迟时，使用负数，标识执行方式为scheduleAtFixedRate。
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
if (delay <= 0)
throw new IllegalArgumentException();
ScheduledFutureTask<Void> sft =
new ScheduledFutureTask<Void>(command,
null,
triggerTime(initialDelay, unit),
unit.toNanos(-delay));
RunnableScheduledFuture<Void> t = decorateTask(command, sft);
sft.outerTask = t;
delayedExecute(t);
return t;
}ScheduledFutureTask并没有设置单独的字段用于标识执行类型，而是通过period字段的正负号和是否为0表示执行方式：
scheduleAtFixedRate() / scheduleWithFixedDelay()执行的主要区别在于设置下一次执行时间的策略不同，而执行时间通过ScheduledFutureTask的time字段保存，通过ScheduledFutureTask#setNextRunTime()进行设置，代码如下：
private void setNextRunTime() {
long p = period;
// 如果是fixed-rate执行方式：下一次执行时间 = 上一次执行时间 + period
if (p > 0)
time += p;
// 如果是fixed-delay执行方式：下一次执行时间 = now() + period
else
time = triggerTime(-p);
}DelayedWorkQueue是专门存放RunnableScheduledFuture和ScheduledFutureTask对象的优先队列，底层基于最小二叉堆实现，为了能够提升任务的查找和删除效率，ScheduledFutureTask中增加了一个heapIndex的成员变量，用于存放任务在堆数组中的索引位置，当需要查找或者删除某个特定的任务时，直接根据任务的heapIndex访问堆数组中的元素。任务是否到达执行时间的判断逻辑均在DelayedWorkQueue中实现。
/**
* 存放堆的数组，初始化大小为16
*/
private RunnableScheduledFuture<?>[] queue =
new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
/**
* 队列中的任务个数
*/
private int size = 0;
/**
* 保证队列操作的锁
*/
private final ReentrantLock lock = new ReentrantLock();
/**
* 存放用于等待任务的leader线程引用；
* 为降低性能消耗，同一时间并不需要所有线程都轮询等待任务到达执行时间；
* 只需要一个leader线程负责轮询等待即可；
*/
private Thread leader = null;
/**
* Condition signalled when a newer task becomes available at the
* head of the queue or a new thread may need to become leader.
*/
private final Condition available = lock.newCondition();
与PriorityQueue的实现不同，DelayedWorkQueue涉及到多线程访问，因而需要保证线程同步测正确性，故使用ReentrantLock来控制操作的原子性，同时使用Condition来协调线程的执行；
为了方便在DelayedWorkQueue中查找和删除任务，ScheduledFutureTask有一个heapIndex用于存放任务在堆数组中的索引位置。每当任务在队列中的位置改变时，需要同步更新任务的heapIndex。
private void setIndex(RunnableScheduledFuture<?> f, int idx) {
if (f instanceof ScheduledFutureTask)
((ScheduledFutureTask)f).heapIndex = idx;
}上浮、下沉操作的实现与PriorityQueue实现相似，只多了更新索引位置的操作，且需要在加锁的环境下调用。
/**
* 上浮操作
*/
private void siftUp(int k, RunnableScheduledFuture<?> key) {
while (k > 0) {
int parent = (k - 1) >>> 1;
RunnableScheduledFuture<?> e = queue[parent];
if (key.compareTo(e) >= 0)
break;
queue[k] = e;
// 更新父节点索引位置
setIndex(e, k);
k = parent;
}
queue[k] = key;
// 更新当前节点索引位置
setIndex(key, k);
}
/**
* 下沉操作
*/
private void siftDown(int k, RunnableScheduledFuture<?> key) {
int half = size >>> 1;
while (k < half) {
int child = (k << 1) + 1;
RunnableScheduledFuture<?> c = queue[child];
int right = child + 1;
if (right < size && c.compareTo(queue[right]) > 0)
c = queue[child = right];
if (key.compareTo(c) <= 0)
break;
queue[k] = c;
// 更新子节点索引
setIndex(c, k);
k = child;
}
queue[k] = key;
// 更新当前节点索引
setIndex(key, k);
} 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);
}
// queue[0] == e包含两种情况：
// 1）e为队列中的第一个元素；
// 2）e为队列中最近要执行的任务；
if (queue[0] == e) {
leader = null;
available.signal();
}
} finally {
lock.unlock();
}
return true;
}
public void put(Runnable e) {
offer(e);
}
public boolean add(Runnable e) {
return offer(e);
} /**
* 任务出队后，通过下沉操作使得堆有序；
*/
private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
int s = --size;
RunnableScheduledFuture<?> x = queue[s];
queue[s] = null;
if (s != 0)
siftDown(0, x);
// 出队任务heapIndex设置为-1
setIndex(f, -1);
return f;
}
/**
* 出队，非阻塞
*/
public RunnableScheduledFuture<?> poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
RunnableScheduledFuture<?> first = queue[0];
// 如果队列为空或者没有任务到执行时间，则返回null
if (first == null || first.getDelay(NANOSECONDS) > 0)
return null;
else
// 执行下沉操作，返回队首任务
return finishPoll(first);
} finally {
lock.unlock();
}
}
/**
* 出队，阻塞线程，直到有任务返回
*/
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)
return finishPoll(first);
// 如果未到达时间点
first = null; // don't retain ref while waiting
// 如果已经有leader线程等待任务，则阻塞当前线程
if (leader != null)
available.await();
// 如果没有leader线程，则设置当前线程为leader线程，轮询等待任务到达执行时间点
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();
}
}
/**
* 出队，阻塞线程，直到有任务返回或者超时
*/
public RunnableScheduledFuture<?> poll(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
// 加锁，允许响应中断
lock.lockInterruptibly();
try {
for (;;) {
RunnableScheduledFuture<?> first = queue[0];
// 如果队列为空
if (first == null) {
// 如果到达超时时间，则返回null
if (nanos <= 0)
return null;
// 未到达超时时间，则等待超时
else
nanos = available.awaitNanos(nanos);
} else {
long delay = first.getDelay(NANOSECONDS);
// 如果队首任务到达执行时间，则立即返回任务；
if (delay <= 0)
return finishPoll(first);
// 如果未到达执行时间，且超时，则返回null；
if (nanos <= 0)
return null;
// 如果未到达执行时间，且没有超时
first = null; // don't retain ref while waiting
// 如果已经有leader线程，或者超时时间小于第一个任务执行时间，
// 则阻塞当前线程直至超时
if (nanos < delay || leader != null)
nanos = available.awaitNanos(nanos);
// 如果没有leader线程，或者没有超时且没有任务到达时间点；
// 则阻塞等待任务到达执行时间点
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
long timeLeft = available.awaitNanos(delay);
nanos -= delay - timeLeft;
} finally {
// 如果当前线程是leader线程，则取消其leader属性
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// 唤醒一个线程，确保至少有一个线程未被阻塞
if (leader == null && queue[0] != null)
available.signal();
lock.unlock();
}
} /**
* 查找一个任务的index，如果未找到，则返回-1
*/
private int indexOf(Object x) {
if (x != null) {
// 如果是ScheduledFutureTask类型任务，则直接返回heapIndex，效率O(1)
if (x instanceof ScheduledFutureTask) {
int i = ((ScheduledFutureTask) x).heapIndex;
// 检查ScheduledFutureTask是否属于当前pool
if (i >= 0 && i < size && queue[i] == x)
return i;
}
// 如果是RunnableScheduledFuture类型任务，则遍历查找，效率O(1)
else {
for (int i = 0; i < size; i++)
if (x.equals(queue[i]))
return i;
}
}
return -1;
}
/**
* 查找任务
*/
public boolean contains(Object x) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return indexOf(x) != -1;
} finally {
lock.unlock();
}
}
/**
* 从队列中删除任务，用于取消任务的场景
*/
public boolean remove(Object x) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 查找任务索引
int i = indexOf(x);
// 未找到任务，则不执行删除操作，返回false
if (i < 0)
return false;
// 清空任务索引信息，减小任务队列size
// 使用队尾任务替换现有任务索引位置，然后通过下沉、上浮操作找到合适位置
setIndex(queue[i], -1);
int s = --size;
RunnableScheduledFuture<?> replacement = queue[s];
queue[s] = null;
if (s != i) {
siftDown(i, replacement);
// 如果任务未下沉，则执行上浮操作
if (queue[i] == replacement)
siftUp(i, replacement);
}
return true;
} finally {
lock.unlock();
}
} 通过上面代码我们总结DelayedWorkQueue的实现原理：
1）基于最小二叉堆实现的优先队列，根据ScheduledFutureTask.compareTo方法比较任务执行时间，使得最近要执行的任务位于队首；
2）任务出队时，通过轮询判断任务是否到达执行时间点，ScheduledFutureTask实现了Delayed接口，通过getDelay方法能够获取到任务还有多长时间执行；
3）当队列中所有任务都没有到达执行时间时，队列中会维持一个leader线程，用于轮询等待队首任务，其余线程均await()。
4）ScheduledFutureTask增加heapIndex属性，用于标记任务在堆数组中的索引，从而便于任务的快速查找（是否存在）与取消（删除）；
任务的取消通过ScheduledFutureTask.cancel()方法实现，该方法调用ThreadPoolExecutor.cancel()，在取消任务后，判断是否需要从阻塞队列中移除任务。其中removeOnCancel参数通过setRemoveOnCancelPolicy()设置。之所以要在取消任务后移除阻塞队列中任务，是为了防止队列中积压大量已被取消的任务。
public boolean cancel(boolean mayInterruptIfRunning) {
// 调用ThreadPoolExecutor.cancel方法取消任务
boolean cancelled = super.cancel(mayInterruptIfRunning);
// 从阻塞队列中移除任务
if (cancelled && removeOnCancel && heapIndex >= 0)
remove(this);
return cancelled;
}ScheduledThreadPoolExecutor的shutdown() / shutdownNow()方法均调用ThreadPoolExecutor的相应方法实现。同时，ScheduledThreadPoolExecutor实现了ThreadPoolExecutor的onShutdown()用于在shutdown()执行过程中取消任务执行。
此处涉及2个参数：
@Override void onShutdown() {
BlockingQueue<Runnable> q = super.getQueue();
boolean keepDelayed = getExecuteExistingDelayedTasksAfterShutdownPolicy();
boolean keepPeriodic = getContinueExistingPeriodicTasksAfterShutdownPolicy();
// 如果设置不执行队列中的任务，则取消队列中所有任务，清空队列；
if (!keepDelayed && !keepPeriodic) {
for (Object e : q.toArray())
if (e instanceof RunnableScheduledFuture<?>)
((RunnableScheduledFuture<?>) e).cancel(false);
q.clear();
}
// 如果设置执行队列中的任务：周期性的或者单次的任务
else {
// 遍历队列中的任务，逐个取消并删除不需要执行的任务
// 先拷贝到数组再遍历，防止遍历时队列元素更新，导致异常；
for (Object e : q.toArray()) {
if (e instanceof RunnableScheduledFuture) {
RunnableScheduledFuture<?> t = (RunnableScheduledFuture<?>)e;
if ((t.isPeriodic() ? !keepPeriodic : !keepDelayed) || t.isCancelled()) {
if (q.remove(t))
t.cancel(false);
}
}
}
}
tryTerminate();
}
微信关注我们
转载内容版权归作者及来源网站所有！
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。
马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。
为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。
Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。
扫码在手机上查看文章
扫描二维码，手机阅读更方便
有任何问题或合作意向欢迎联系我们
Email: 99873273@qq.com
QQ: 99873273