【高并发】深度解析ScheduledThreadPoolExecutor类的源代码( 三 )


overflowFree方法overflowFree方法的源代码如下所示 。
private long overflowFree(long delay) { //获取队列中的节点 Delayed head = (Delayed) super.getQueue().peek(); //获取的节点不为空,则进行后续处理 if (head != null) {//从队列节点中获取延迟时间long headDelay = head.getDelay(NANOSECONDS);//如果从队列中获取的延迟时间小于0,并且传递的delay//值减去从队列节点中获取延迟时间小于0if (headDelay < 0 && (delay - headDelay < 0))//将delay的值设置为Long.MAX_VALUE + headDelaydelay = Long.MAX_VALUE + headDelay; } //返回延迟时间 return delay;}通过对overflowFree方法的源码分析,可以看出overflowFree方法本质上就是为了限制队列中的所有节点的延迟时间在Long.MAX_VALUE值之内,防止在ScheduledFutureTask类中的compareTo方法中溢出 。
ScheduledFutureTask类中的compareTo方法的源码如下所示 。
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;}compareTo方法的主要作用就是对各延迟任务进行排序,距离下次执行时间靠前的任务就排在前面 。
delayedExecute方法delayedExecute方法是ScheduledThreadPoolExecutor类中延迟执行任务的方法,源代码如下所示 。
private void delayedExecute(RunnableScheduledFuture<?> task) { //如果当前线程池已经关闭 //则执行线程池的拒绝策略 if (isShutdown())reject(task); //线程池没有关闭 else {//将任务添加到阻塞队列中super.getQueue().add(task);//如果当前线程池是SHUTDOWN状态//并且当前线程池状态下不能执行任务//并且成功从阻塞队列中移除任务if (isShutdown() &&!canRunInCurrentRunState(task.isPeriodic()) &&remove(task))//取消任务的执行,但不会中断执行中的任务task.cancel(false);else//调用ThreadPoolExecutor类中的ensurePrestart()方法ensurePrestart(); }}可以看到在delayedExecute方法内部调用了canRunInCurrentRunState方法,canRunInCurrentRunState方法的源码实现如下所示 。
boolean canRunInCurrentRunState(boolean periodic) { return isRunningOrShutdown(periodic ? continueExistingPeriodicTasksAfterShutdown : executeExistingDelayedTasksAfterShutdown);}可以看到canRunInCurrentRunState方法的逻辑比较简单,就是判断线程池当前状态下能够执行任务 。
另外,在delayedExecute方法内部还调用了ThreadPoolExecutor类中的ensurePrestart()方法,接下来,我们看下ThreadPoolExecutor类中的ensurePrestart()方法的实现,如下所示 。
【【高并发】深度解析ScheduledThreadPoolExecutor类的源代码】void ensurePrestart() { int wc = workerCountOf(ctl.get()); if (wc < corePoolSize)addWorker(null, true); else if (wc == 0)addWorker(null, false);}在ThreadPoolExecutor类中的ensurePrestart()方法中,首先获取当前线程池中线程的数量,如果线程数量小于corePoolSize则调用addWorker方法传递null和true,如果线程数量为0,则调用addWorker方法传递null和false 。
关于addWork()方法的源码解析,大家可以参考【高并发专题】中的《高并发之——通过ThreadPoolExecutor类的源码深度解析线程池执行任务的核心流程》一文,这里,不再赘述 。
reExecutePeriodic方法reExecutePeriodic方法的源代码如下所示 。
void reExecutePeriodic(RunnableScheduledFuture<?> task) { //线程池当前状态下能够执行任务 if (canRunInCurrentRunState(true)) {//将任务放入队列super.getQueue().add(task);//线程池当前状态下不能执行任务,并且成功移除任务if (!canRunInCurrentRunState(true) && remove(task))//取消任务task.cancel(false);else//调用ThreadPoolExecutor类的ensurePrestart()方法ensurePrestart(); }}

经验总结扩展阅读