ScheduledThreadPoolExecutor中定时周期任务的实…
2020-01-07 08:33:46来源:博客园 阅读 ()
ScheduledThreadPoolExecutor中定时周期任务的实现源码分析
ScheduledThreadPoolExecutor是一个定时任务线程池,相比于ThreadPoolExecutor最大的不同在于其阻塞队列的实现
首先看一下其构造方法:
1 public ScheduledThreadPoolExecutor(int corePoolSize, 2 ThreadFactory threadFactory, 3 RejectedExecutionHandler handler) { 4 super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, 5 new DelayedWorkQueue(), threadFactory, handler); 6 }
ScheduledThreadPoolExecutor是继承自ThreadPoolExecutor的,可以看到这里实际上调用了ThreadPoolExecutor的构造方法,而最大的不同在于这里使用了默认的DelayedWorkQueue“阻塞队列”,这是后续能够实现定时任务的关键
在ScheduledThreadPoolExecutor中使用scheduleWithFixedDelay或者scheduleAtFixedRate方法来完成定时周期任务
以scheduleWithFixedDelay为例
scheduleWithFixedDelay方法:
1 public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, 2 long initialDelay, 3 long delay, 4 TimeUnit unit) { 5 if (command == null || unit == null) 6 throw new NullPointerException(); 7 if (delay <= 0) 8 throw new IllegalArgumentException(); 9 ScheduledFutureTask<Void> sft = 10 new ScheduledFutureTask<Void>(command, 11 null, 12 triggerTime(initialDelay, unit), 13 unit.toNanos(-delay)); 14 RunnableScheduledFuture<Void> t = decorateTask(command, sft); 15 sft.outerTask = t; 16 delayedExecute(t); 17 return t; 18 }
这里首先会将我们的任务包装成ScheduledFutureTask
(这里的delay在传入ScheduledFutureTask的构造方法时变为了负的,这是和scheduleAtFixedRate方法唯一不一样的地方)
ScheduledFutureTask方法:
1 private void delayedExecute(RunnableScheduledFuture<?> task) { 2 if (isShutdown()) 3 reject(task); 4 else { 5 super.getQueue().add(task); 6 if (isShutdown() && 7 !canRunInCurrentRunState(task.isPeriodic()) && 8 remove(task)) 9 task.cancel(false); 10 else 11 ensurePrestart(); 12 } 13 }
这里不同于ThreadPoolExecutor中的处理,并没有考虑coreSize和maxSize和任务之间的关系,而是直接将任务提交到阻塞队列DelayedWorkQueue中
DelayedWorkQueue的add方法:
1 public boolean add(Runnable e) { 2 return offer(e); 3 } 4 5 public boolean offer(Runnable x) { 6 if (x == null) 7 throw new NullPointerException(); 8 RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x; 9 final ReentrantLock lock = this.lock; 10 lock.lock(); 11 try { 12 int i = size; 13 if (i >= queue.length) 14 grow(); 15 size = i + 1; 16 if (i == 0) { 17 queue[0] = e; 18 setIndex(e, 0); 19 } else { 20 siftUp(i, e); 21 } 22 if (queue[0] == e) { 23 leader = null; 24 available.signal(); 25 } 26 } finally { 27 lock.unlock(); 28 } 29 return true; 30 }
实际上调用了offer方法,从这里就可以看出这个“阻塞队列”的不同之处
DelayedWorkQueue中有这些成员:
1 private static final int INITIAL_CAPACITY = 16; 2 private RunnableScheduledFuture<?>[] queue = 3 new RunnableScheduledFuture<?>[INITIAL_CAPACITY]; 4 private int size = 0; 5 private Thread leader = null;
在DelayedWorkQueue内部维护的是queue这个初始大小16的数组,其实就是一个小根堆
回到offer方法
由于是在多线程环境,这里的操作使用了重入锁保证原子性
若是在size大于数组的长度情况下,就需要调用grow方法来扩容
grow方法:
1 private void grow() { 2 int oldCapacity = queue.length; 3 int newCapacity = oldCapacity + (oldCapacity >> 1); // grow 50% 4 if (newCapacity < 0) // overflow 5 newCapacity = Integer.MAX_VALUE; 6 queue = Arrays.copyOf(queue, newCapacity); 7 }
可以看到这是一个非常简单的扩容机制,申请一个1.5倍大小的新数组,再将原来的数据copy上去
回到offer方法,在调整完容量后,就需要进行数据的插入,使其形成一个小根堆
可以看到,在if-else判断中,首先检查是不是第一个元素,若是第一个,则直接放入数组,同时调用
setIndex方法,和任务关联
setIndex方法:
1 private void setIndex(RunnableScheduledFuture<?> f, int idx) { 2 if (f instanceof ScheduledFutureTask) 3 ((ScheduledFutureTask)f).heapIndex = idx; 4 }
这个方法很简单,将下标关联到之前包装好的任务ScheduledFutureTask中
若不是第一个元素,则需要调用siftUp,进行小根堆的调整
siftUp方法:
1 private void siftUp(int k, RunnableScheduledFuture<?> key) { 2 while (k > 0) { 3 int parent = (k - 1) >>> 1; 4 RunnableScheduledFuture<?> e = queue[parent]; 5 if (key.compareTo(e) >= 0) 6 break; 7 queue[k] = e; 8 setIndex(e, k); 9 k = parent; 10 } 11 queue[k] = key; 12 setIndex(key, k); 13 }
因为小根堆实际上就是一个二叉树,利用二叉树的性质根据当前要插入节点的下标,得到其父节点的下标parent ,再和父节点的RunnableScheduledFuture对象进行compareTo的比较(RunnableScheduledFuture继承了Comparable接口)
compareTo的实现:
1 public int compareTo(Delayed other) { 2 if (other == this) // compare zero if same object 3 return 0; 4 if (other instanceof ScheduledFutureTask) { 5 ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other; 6 long diff = time - x.time; 7 if (diff < 0) 8 return -1; 9 else if (diff > 0) 10 return 1; 11 else if (sequenceNumber < x.sequenceNumber) 12 return -1; 13 else 14 return 1; 15 } 16 long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS); 17 return (diff < 0) ? -1 : (diff > 0) ? 1 : 0; 18 }
这里的逻辑比较简单,只需要看第二个if
在前面ScheduledFutureTask包装我们的任务的时候,其构造方法如下:
1 ScheduledFutureTask(Runnable r, V result, long ns, long period) { 2 super(r, result); 3 this.time = ns; 4 this.period = period; 5 this.sequenceNumber = sequencer.getAndIncrement(); 6 }
这里的time 也就是initialDelay,period 就是-delay,sequenceNumber 是一个全局自增的序列号
那么在上面的compareTo方法中,就首先根据子节点的initialDelay和父节点的initialDelay比较
若是子节点小于父节点,返回-1,子节点大于父节点返回1
若是相等,则根据序列号比较,序列号小的返回-1
回到siftUp方法,通过compareTo方法,若是大于等于0,就说明子节点大于父节点,不需要做调整,结束循环
若是小于0,说明子节点小于父节点,那么就需要将父节点先交换到当前位置,再将k变成parent,在下一次循环时,就会找parent的parent,重复上述操作,直至构成小根堆
最后将要插入的节点放入queue中合适的位置
那么在后续的任务添加中,就会根据任务的initialDelay,以及创建时间,构建一个小根堆
回到offer方法,在小根堆中插入完节点后,若是第一次插入, 将leader(Thread对象)置为null,利用available(Condition对象)唤醒Lock 的AQS上的阻塞
DelayedWorkQueue的add到此结束,回到delayedExecute方法中,在完成向阻塞队列添加任务后,发现线程池中并没有一个worker在工作,接下来的工作就由ThreadPoolExecutor的ensurePrestart方法实现:
1 void ensurePrestart() { 2 int wc = workerCountOf(ctl.get()); 3 if (wc < corePoolSize) 4 addWorker(null, true); 5 else if (wc == 0) 6 addWorker(null, false); 7 }
可以看到这里根据ctl的取值,与corePoolSize比较,调用了线程池的addWorker方法,那么实际上也就是通过这里开启了线程池的worker来进行工作
来看看在worker的轮询中发生了什么:
1 final void runWorker(Worker w) { 2 Thread wt = Thread.currentThread(); 3 Runnable task = w.firstTask; 4 w.firstTask = null; 5 w.unlock(); // allow interrupts 6 boolean completedAbruptly = true; 7 try { 8 while (task != null || (task = getTask()) != null) { 9 w.lock(); 10 // If pool is stopping, ensure thread is interrupted; 11 // if not, ensure thread is not interrupted. This 12 // requires a recheck in second case to deal with 13 // shutdownNow race while clearing interrupt 14 if ((runStateAtLeast(ctl.get(), STOP) || 15 (Thread.interrupted() && 16 runStateAtLeast(ctl.get(), STOP))) && 17 !wt.isInterrupted()) 18 wt.interrupt(); 19 try { 20 beforeExecute(wt, task); 21 Throwable thrown = null; 22 try { 23 task.run(); 24 } catch (RuntimeException x) { 25 thrown = x; throw x; 26 } catch (Error x) { 27 thrown = x; throw x; 28 } catch (Throwable x) { 29 thrown = x; throw new Error(x); 30 } finally { 31 afterExecute(task, thrown); 32 } 33 } finally { 34 task = null; 35 w.completedTasks++; 36 w.unlock(); 37 } 38 } 39 completedAbruptly = false; 40 } finally { 41 processWorkerExit(w, completedAbruptly); 42 } 43 }
可以看到在ThreadPoolExecutor的worker轮询线程中,会通过getTask方法,不断地从阻塞队列中获取任务
getTask方法:
1 private Runnable getTask() { 2 boolean timedOut = false; // Did the last poll() time out? 3 4 for (;;) { 5 int c = ctl.get(); 6 int rs = runStateOf(c); 7 8 // Check if queue empty only if necessary. 9 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 10 decrementWorkerCount(); 11 return null; 12 } 13 14 int wc = workerCountOf(c); 15 16 // Are workers subject to culling? 17 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; 18 19 if ((wc > maximumPoolSize || (timed && timedOut)) 20 && (wc > 1 || workQueue.isEmpty())) { 21 if (compareAndDecrementWorkerCount(c)) 22 return null; 23 continue; 24 } 25 26 try { 27 Runnable r = timed ? 28 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : 29 workQueue.take(); 30 if (r != null) 31 return r; 32 timedOut = true; 33 } catch (InterruptedException retry) { 34 timedOut = false; 35 } 36 } 37 }
可以看到在这个方法中,在一系列的参数检查并设置完毕后,会通过workQueue的poll或者take方法来获取所需的任务
其中poll方法是在设置了超时时间的情况下进行获取,take则不带有超时时间
以take为例
DelayedWorkQueue的take方法:
1 public RunnableScheduledFuture<?> take() throws InterruptedException { 2 final ReentrantLock lock = this.lock; 3 lock.lockInterruptibly(); 4 try { 5 for (;;) { 6 RunnableScheduledFuture<?> first = queue[0]; 7 if (first == null) 8 available.await(); 9 else { 10 long delay = first.getDelay(NANOSECONDS); 11 if (delay <= 0) 12 return finishPoll(first); 13 first = null; // don't retain ref while waiting 14 if (leader != null) 15 available.await(); 16 else { 17 Thread thisThread = Thread.currentThread(); 18 leader = thisThread; 19 try { 20 available.awaitNanos(delay); 21 } finally { 22 if (leader == thisThread) 23 leader = null; 24 } 25 } 26 } 27 } 28 } finally { 29 if (leader == null && queue[0] != null) 30 available.signal(); 31 lock.unlock(); 32 } 33 }
在for循环中首先取出数组中的第一个元素,也就是生成的小根堆中最小的那一个
得到first后,若是first为null,则说明当前没有可执行的任务,则使用available这个Condition对象,将AQS阻塞起来,等待下次任务创建时再通过前面提到的available唤醒阻塞
若是first存在,则通过getDelay方法获取时间间隔
getDelay方法:
1 public long getDelay(TimeUnit unit) { 2 return unit.convert(time - now(), NANOSECONDS); 3 }
这个方法就是用time减去当前时间now,得到的一个纳秒级时间差值
而time是在ScheduledFutureTask执行构造方法时,通过triggerTime方法,使用initialDelay进行计算出来的
triggerTime方法:
1 private long triggerTime(long delay, TimeUnit unit) { 2 return triggerTime(unit.toNanos((delay < 0) ? 0 : delay)); 3 } 4 5 long triggerTime(long delay) { 6 return now() + 7 ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay)); 8 } 9 10 private long overflowFree(long delay) { 11 Delayed head = (Delayed) super.getQueue().peek(); 12 if (head != null) { 13 long headDelay = head.getDelay(NANOSECONDS); 14 if (headDelay < 0 && (delay - headDelay < 0)) 15 delay = Long.MAX_VALUE + headDelay; 16 } 17 return delay; 18 }
可以看到time在这里实际上就是通过initialDelay加上当时设置的纳秒级时间组成的
其中overflowFree是为了防止Long类型的溢出做了一次计算,后边再说
所以take方法中,通过getDelay方法得到的是一个时间差,若是时间差小于等于0,则说明任务到了该执行的时候了,此时调用finishPoll
finishPoll方法:
1 private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) { 2 int s = --size; 3 RunnableScheduledFuture<?> x = queue[s]; 4 queue[s] = null; 5 if (s != 0) 6 siftDown(0, x); 7 setIndex(f, -1); 8 return f; 9 }
这个方法的逻辑还是比较简单的,就是一个简单的小根堆重新调整的操作,由于f需要被取出,此时利用最后一个元素,完成一次自上向下的调整(生成时是自下向上)
siftDown方法和siftUp类似:
1 private void siftDown(int k, RunnableScheduledFuture<?> key) { 2 int half = size >>> 1; 3 while (k < half) { 4 int child = (k << 1) + 1; 5 RunnableScheduledFuture<?> c = queue[child]; 6 int right = child + 1; 7 if (right < size && c.compareTo(queue[right]) > 0) 8 c = queue[child = right]; 9 if (key.compareTo(c) <= 0) 10 break; 11 queue[k] = c; 12 setIndex(c, k); 13 k = child; 14 } 15 queue[k] = key; 16 setIndex(key, k); 17 }
由二叉树性质half 保证只操作到倒数第二层
在循环中,首先根据k(当前也就是根节点),得到其左右孩子的下标
若是右孩子存在,那么就用左孩子和右孩子比较,选出最下的哪一个作为child
若是右孩子不存在,则直接使用左孩子作为child
当选出child后,再和待插入的元素key比较
若是key小,则结束循环,直接将key插入k所在位置
若不是,则将当前child所在元素放在k所在位置,然后从child位置继续开始向下寻找,直到找到一个大于key或者遍历完毕
这样自上向下的将当前堆又调整成了小根堆,以后的定时周期任务都是以这种方式来调用的
看到这ScheduledThreadPoolExecutor的定时周期任务已经基本理解了,只不过还存在一个问题,当执行周期任务,会从小根堆取出,那么该任务下一次的执行时间何时更新到小根堆?
回到ThreadPoolExecutor的worker的runWorker方法中,在调用完getTask方法后,在进行完一系列完全检查后,会直接调用task的run方法,而此时的task是经过之前ScheduledFutureTask包装的
ScheduledFutureTask的run方法:
1 public void run() { 2 boolean periodic = isPeriodic(); 3 if (!canRunInCurrentRunState(periodic)) 4 cancel(false); 5 else if (!periodic) 6 ScheduledFutureTask.super.run(); 7 else if (ScheduledFutureTask.super.runAndReset()) { 8 setNextRunTime(); 9 reExecutePeriodic(outerTask); 10 } 11 }
若是设置了周期任务(period不为0),那么isPeriodic方法为true
逻辑上就会执行runAndReset方法,这个方法内部就会调用我们传入的Runnable的run方法,从而真正地执行我们的任务
在执行完毕后,可以看到调用了setNextRunTime方法
setNextRunTime方法:
1 private void setNextRunTime() { 2 long p = period; 3 if (p > 0) 4 time += p; 5 else 6 time = triggerTime(-p); 7 }
这里就很简单,利用当前time和period计算出下一次的time
由于scheduleWithFixedDelay和scheduleAtFixedRate之前所说的不一样之处,在这里就得到了体现
因为scheduleAtFixedRate的period是大于0的,所以scheduleAtFixedRate计算出来的时间间隔就是initialDelay + n*period的这种形式,那么其执行就会有固定的时间点,不过这还是要取决于任务的执行时间,若是任务的执行时间大于时间间隔,那么当上一次任务执行完毕,就会立刻执行,而不是等到时间点到了,若是任务的执行时间小于时间间隔,那么毫无疑问就需要等到时间点到了才执行下一次的任务
由于scheduleWithFixedDelay的period是小于0的,所以需要执行triggerTime
triggerTime方法:
1 long triggerTime(long delay) { 2 return now() + 3 ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay)); 4 }
可以看到若是不存在Long类型的溢出问题,那么下一次的时间就等于当前时间加时间间隔,所以说scheduleWithFixedDelay的不同之处在于其算上了任务的实际执行时间
若是存在Long类型的溢出问题时
在overflowFree中:
1 private long overflowFree(long delay) { 2 Delayed head = (Delayed) super.getQueue().peek(); 3 if (head != null) { 4 long headDelay = head.getDelay(NANOSECONDS); 5 if (headDelay < 0 && (delay - headDelay < 0)) 6 delay = Long.MAX_VALUE + headDelay; 7 } 8 return delay; 9 }
首先通过peek得到队列中的第一个元素,若是不存在,则直接返回delay
若是存在,通过getDelay得到headDelay
这里就会存在两情况
任务还没达到执行时间,则headDelay 大于零
任务达到执行时间,但却由于之前的任务还没执行完毕,遭到了延时,headDelay 小于0
所以这次的计算就是将headDelay这部分超时时间减去,以防止后续影响compareTo的比较,从而引起offer顺序的错误
(只不过这种情况正常不会遇见。。。)
在计算完成下一次的运行时间后
调用reExecutePeriodic方法:
1 void reExecutePeriodic(RunnableScheduledFuture<?> task) { 2 if (canRunInCurrentRunState(true)) { 3 super.getQueue().add(task); 4 if (!canRunInCurrentRunState(true) && remove(task)) 5 task.cancel(false); 6 else 7 ensurePrestart(); 8 } 9 }
其中传入的这个task(outerTask)其实就是当前执行完毕的这个任务,
可以看到这里canRunInCurrentRunState成立的情况下,就会通过
getQueue得到阻塞队列,再次通过DelayedWorkQueue的add方法将其加入到小根堆中,只不过这时的time发生了变化
若是情况正常,则继续通过ThreadPoolExecutor的ensurePrestart方法,调度worker的工作
这样定时周期任务就能正常执行
ScheduledThreadPoolExecutor分析到此结束
原文链接:https://www.cnblogs.com/a526583280/p/12162173.html
如有疑问请与原作者联系
标签:
版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有
上一篇:使用DOM4J生成XML文档
下一篇:使用DOM解析XML文档
- Spring Boot 实现定时任务的 4 种方式 2020-06-10
- SpringBoot通过web页面动态控制定时任务的启动、停止、创建 2020-06-09
- SpringBoot定时任务如何正确运用?案例详解 2020-06-05
- java学习心得——Quartz 自定义定时器的操作 2020-05-20
- SpringBoot集成Quartz实现定时任务 2020-05-06
IDC资讯: 主机资讯 注册资讯 托管资讯 vps资讯 网站建设
网站运营: 建站经验 策划盈利 搜索优化 网站推广 免费资源
网络编程: Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它
服务器技术: Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护
软件技巧: 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷
网页制作: FrontPages Dreamweaver Javascript css photoshop fireworks Flash