亲宝软件园·资讯

展开

ScheduledThreadPoolExecutor中定时周期任务的实现源码分析

松饼人 人气:2

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分析到此结束

加载全部内容

相关教程
猜你喜欢
用户评论