简介

它是一个线程池执行器（ThreadPoolExecutor）,在给定的延迟（delay）后执行。在多线程或者对灵活性有要求的环境下，要优于java.util.Timer。

提交的任务在执行之前支持取消，默认情况下，在延迟到来之前，不会自动从队列中删除，但可以设置，使其立刻从队列中移除。

有两种模式，固定频率（scheduleAtFixedRate）和固定延迟（scheduleWithFixedDelay），不管哪种模式，同一个任务不会被叠加执行，即便是不同的线程执行同一个任务。

继承ThreadPoolExecutor，维护一个固定大小的线程池和一个无界延迟队列（delay queue）。

ScheduledFutureTask，用来描述要执行的任务，DelayedWorkQueue，则是装在这些任务的delay queue.

固定频率

一个任务，从第一次开始执行的时间点开始，每隔一定的时间执行一次，如果执行的时间大于间隔时间，则要等这次执行结束，再执行下一次。

如上图所示，蓝色表示任务执行，白色表示间隔时间。

固定延迟

一个任务，每一次执行结束之后，延迟一定的时间，执行下一次。

如上图所示，蓝色表示任务执行，白色表示间隔时间。

源码分析

属性

     private volatile boolean continueExistingPeriodicTasksAfterShutdown; // shut down之后，是否取消period任务

     private volatile boolean executeExistingDelayedTasksAfterShutdown = true; // shut down之后，是否取消non-period任务

     private volatile boolean removeOnCancel = false; // cancel后，是否从队列里移除此任务

     private static final AtomicLong sequencer = new AtomicLong(); // 给任务编号

ScheduledFutureTask

属性

         private final long sequenceNumber; // 序列编号

         private long time; // 执行时间

         private final long period; // 周期，正值：固定频率；负值：固定延迟；0：不重复执行

         RunnableScheduledFuture<V> outerTask = this; // 实际任务

         int heapIndex; // 堆索引

构造方法

         ScheduledFutureTask(Runnable r, V result, long ns) {

             super(r, result);

             this.time = ns;

             this.period = 0; // 不重复执行

             this.sequenceNumber = sequencer.getAndIncrement();

         }

         ScheduledFutureTask(Runnable r, V result, long ns, long period) {

             super(r, result);

             this.time = ns;

             this.period = period;

             this.sequenceNumber = sequencer.getAndIncrement();

         }

         ScheduledFutureTask(Callable<V> callable, long ns) {

             super(callable);

             this.time = ns;

             this.period = 0;

             this.sequenceNumber = sequencer.getAndIncrement();

         }

关键方法

getDelay(TimeUnit)

         public long getDelay(TimeUnit unit) {

             return unit.convert(time - now(), NANOSECONDS);

         }

compareTo(Delayed other)

         public int compareTo(Delayed other) { // 根据延迟比较元素，在延迟队列中，延迟越小越靠前，延迟最小的在队首，最先出队被执行

             if (other == this)

                 return 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;

                 else

                     return 1;

             }

             long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);

             return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;

         }

setNextRunTime()

         private void setNextRunTime() { // 设置下一次执行时间

             long p = period;

             if (p > 0) // 固定频率，从第一次时间点，每次加period

                 time += p;

             else

                 time = triggerTime(-p); // 固定延迟，每次执行结束后，加period作为下一次执行时间

         }

cancel(boolean mayInterruptIfRunning)

         public boolean cancel(boolean mayInterruptIfRunning) { // 取消任务

             boolean cancelled = super.cancel(mayInterruptIfRunning);

             if (cancelled && removeOnCancel && heapIndex >= 0)

                 remove(this);

             return cancelled;

         }

run()

         public void run() { // 执行任务

             boolean periodic = isPeriodic();

             if (!canRunInCurrentRunState(periodic))

                 cancel(false);

             else if (!periodic)

                 ScheduledFutureTask.super.run(); // 单次执行

             else if (ScheduledFutureTask.super.runAndReset()) { // 周期执行，runAndReset

                 setNextRunTime(); // 设置下次执行时间

                 reExecutePeriodic(outerTask); // 重新加入队列

             }

         }

构造方法

     public ScheduledThreadPoolExecutor(int corePoolSize) {

         super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue());

     }

     public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory) {

         super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), threadFactory);

     }

     public ScheduledThreadPoolExecutor(int corePoolSize, RejectedExecutionHandler handler) {

         super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), handler);

     }

     public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory,

             RejectedExecutionHandler handler) {

         super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), threadFactory, handler);

     }

关键方法

scheduleAtFixedRate

     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)); // period 大于 0

         RunnableScheduledFuture<Void> t = decorateTask(command, sft);

         sft.outerTask = t;

         delayedExecute(t);

         return t;

     }

scheduleWithFixedDelay

     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)); // -delay 小于 0

         RunnableScheduledFuture<Void> t = decorateTask(command, sft);

         sft.outerTask = t;

         delayedExecute(t);

         return t;

     }

delayedExecute(RunnableScheduledFuture<?> task)

     private void delayedExecute(RunnableScheduledFuture<?> task) {

         if (isShutdown()) // 如果线程池已经shut down，则拒绝任务

             reject(task);

         else {

             super.getQueue().add(task); // 否则，添加任务到延迟队列

             if (isShutdown() && !canRunInCurrentRunState(task.isPeriodic()) && remove(task))

                 task.cancel(false); // 根据run-after-shutdown参数，决定是否取任务

             else

                 ensurePrestart(); // 保证线程启动

         }

     }

reExecutePeriodic(RunnableScheduledFuture<?> task)

     void reExecutePeriodic(RunnableScheduledFuture<?> task) { // 周期性任务重新入队，策略同delayedExecute

         if (canRunInCurrentRunState(true)) {

             super.getQueue().add(task);

             if (!canRunInCurrentRunState(true) && remove(task))

                 task.cancel(false);

             else

                 ensurePrestart();

         }

     }

ensurePrestart()

     void ensurePrestart() {

         int wc = workerCountOf(ctl.get());

         if (wc < corePoolSize)

             addWorker(null, true);

         else if (wc == 0)

             addWorker(null, false);

     }

该方法在ThreadPoolExecutor类，保证线程池中至少有一个活动线程。

triggerTime()

     long triggerTime(long delay) { // 返回延迟动作的触发时间

         return now() + ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));

     }

     private long overflowFree(long delay) { // 处理溢出情况

         Delayed head = (Delayed) super.getQueue().peek();

         if (head != null) {

             long headDelay = head.getDelay(NANOSECONDS);

             if (headDelay < 0 && (delay - headDelay < 0))

                 delay = Long.MAX_VALUE + headDelay;

         }

         return delay;

     }

DelayedWorkQueue

属性

         private static final int INITIAL_CAPACITY = 16; // 初始容量

         private RunnableScheduledFuture<?>[] queue = new RunnableScheduledFuture<?>[INITIAL_CAPACITY]; // 堆，充当优先级队列

         private final ReentrantLock lock = new ReentrantLock(); // 可重入锁

         private int size = 0;

         private Thread leader = null; // 领导者线程

         private final Condition available = lock.newCondition(); // 条件队列

关键方法

以下这些方法的解释可参考前两篇文章，【JUC源码解析】DelayQueue和【JUC源码解析】PriorityBlockingQueue

siftUp

         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);

         }

siftDown

         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);

         }

grow

         private void grow() { // 扩容

             int oldCapacity = queue.length;

             int newCapacity = oldCapacity + (oldCapacity >> 1); // grow 50%

             if (newCapacity < 0) // overflow

                 newCapacity = Integer.MAX_VALUE;

             queue = Arrays.copyOf(queue, newCapacity);

         }

offer

         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);

                 }

                 if (queue[0] == e) {

                     leader = null;

                     available.signal();

                 }

             } finally {

                 lock.unlock();

             }

             return true;

         }

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);

                         if (delay <= 0)

                             return finishPoll(first);

                         first = null; // don't retain ref while waiting

                         if (leader != null)

                             available.await();

                         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();

             }

         }

行文至此结束。

尊重他人的劳动，转载请注明出处：http://www.cnblogs.com/aniao/p/aniao_stpe.html

巴特西

【JUC源码解析】ScheduledThreadPoolExecutor

简介

固定频率

固定延迟

源码分析

属性

ScheduledFutureTask

属性

构造方法

关键方法

getDelay(TimeUnit)

compareTo(Delayed other)

setNextRunTime()

cancel(boolean mayInterruptIfRunning)

run()

构造方法

关键方法

scheduleAtFixedRate

scheduleWithFixedDelay

delayedExecute(RunnableScheduledFuture<?> task)

reExecutePeriodic(RunnableScheduledFuture<?> task)

ensurePrestart()

triggerTime()

DelayedWorkQueue

属性

关键方法

siftUp

siftDown

grow

offer

take

最新文章

热门文章