Jdk1.6 JUC源码解析(19)-ScheduledThreadPoolExecutor

Jdk1.6 JUC源码解析(19)-ScheduledThreadPoolExecutor

作者：大飞

 

功能简介：

• ScheduledThreadPoolExecutor是一种类似Timer的定时器或者说是调度器，和Timer比起来主要有几点好处：1.多线程的定时调度，timer是单线程的，每个timer实例只有一个工作线程。2.由于继承自ThreadPoolExecutor，更具有灵活性和伸缩性。3.没有timer那种线程泄露问题，timer调度的任务如果异常终止，那么整个timer都会被取消，无法执行其他任务。

源码分析：

• ScheduledThreadPoolExecutor继承ThreadPoolExecutor，并实现了ScheduledExecutorService接口，ThreadPoolExecutor之前做过分析，这里就简单看下ScheduledExecutorService接口：

public interface ScheduledExecutorService extends ExecutorService {
    /**
     * 创建并执行一个一次性任务，这个任务过了延迟时间就会被执行。
     */
    public ScheduledFuture<?> schedule(Runnable command,
				       long delay, TimeUnit unit);
    /**
     * 创建并执行一个一次性任务，这个任务过了延迟时间就会被执行。
     */
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
					   long delay, TimeUnit unit);
    /**
     * 创建并执行一个周期性任务，当任务过了给定的初始延迟时间，会第一
     * 次被执行，然后会以给定的周期时间执行。
     * 如果某次执行过程中发生了异常，那么任务就停止了(不会执行下一次任务了)。
     * 如果某次执行时长超过了周期时间，那么下一次任务会延迟启动，不会和当前
     * 任务并行执行。
     */
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
						  long initialDelay,
						  long period,
						  TimeUnit unit);
    /**
     * 创建并执行一个周期性任务，当任务过了给定的初始延迟时间，会第一
     * 次被执行，接下来的任务会在上次任务执行完毕后，延迟给定的时间，
     * 然后再继续执行。
     * 如果某次执行过程中发生了异常，那么任务就停止了(不会执行下一次任务了)。
     */
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
						     long initialDelay,
						     long delay,
						     TimeUnit unit);
}

 

       接来下看下ScheduledThreadPoolExecutor内部的一些属性定义：

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {
    /**
     * 表示是否应该在关闭时取消或者终止周期性任务。
     */
    private volatile boolean continueExistingPeriodicTasksAfterShutdown;
    /**
     * 表示是否应该在关闭时取消非周期性任务。
     */
    private volatile boolean executeExistingDelayedTasksAfterShutdown = true;
    /**
     * 这个序列号的作用是在并列调度(延迟值一样)的情况下保证先入先出的关系。
     */
    private static final AtomicLong sequencer = new AtomicLong(0);
    /** Base of nanosecond timings, to avoid wrapping */
    private static final long NANO_ORIGIN = System.nanoTime();

 

       继续看下ScheduledThreadPoolExecutor的构造方法： 

    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue());
    }

    public ScheduledThreadPoolExecutor(int corePoolSize,
                             ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue(), threadFactory);
    }

    public ScheduledThreadPoolExecutor(int corePoolSize,
                              RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue(), handler);
    }
    public ScheduledThreadPoolExecutor(int corePoolSize,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue(), threadFactory, handler);
    }

       从构造方法可以看出，ScheduledThreadPoolExecutor内部固定使用DelayedWorkQueue做为任务队列，DelayedWorkQueue是啥呢？看下代码：  

    /**
     * An annoying wrapper class to convince javac to use a
     * DelayQueue<RunnableScheduledFuture> as a BlockingQueue<Runnable>
     */
    private static class DelayedWorkQueue
        extends AbstractCollection<Runnable>
        implements BlockingQueue<Runnable> {
        private final DelayQueue<RunnableScheduledFuture> dq = new DelayQueue<RunnableScheduledFuture>();
        ...
        public boolean add(Runnable x) {
	    return dq.add((RunnableScheduledFuture)x);
	}
        public boolean offer(Runnable x) {
	    return dq.offer((RunnableScheduledFuture)x);
	}
        public void put(Runnable x) {
            dq.put((RunnableScheduledFuture)x);
        }
        public boolean offer(Runnable x, long timeout, TimeUnit unit) {
            return dq.offer((RunnableScheduledFuture)x, timeout, unit);
        }

       可见，DelayedWorkQueue内部就是一个DelayQueue，所有方法都由内部的DelayQueue来代理实现，唯一要注意的地方只是"规范"了进入队列的任务必须是RunnableScheduledFuture。再回头看ScheduledThreadPoolExecutor，因为延迟队列是无界的，所以最大线程数量也就没意义了。

 

• 下面从ScheduledExecutorService接口的方法来入手，查看代码实现细节，首先看下schedule方法：

    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        //将任务包装成一个RunnableScheduledFuture
        RunnableScheduledFuture<?> t = decorateTask(command,
            new ScheduledFutureTask<Void>(command, null,
                                          triggerTime(delay, unit)));
        //然后延迟执行这个RunnableScheduledFuture
        delayedExecute(t);
        return t;
    }
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay,
                                           TimeUnit unit) {
        if (callable == null || unit == null)
            throw new NullPointerException();
        RunnableScheduledFuture<V> t = decorateTask(callable,
            new ScheduledFutureTask<V>(callable,
	   			       triggerTime(delay, unit)));
        delayedExecute(t);
        return t;
    }

       两个方法内部基本一致，具体看下将任务包装成RunnableScheduledFuture的过程：

    /**
     * 返回一个延迟动作的触发时间。
     */
    private long triggerTime(long delay, TimeUnit unit) {
         //内部要转成纳秒。
         return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
    }
    /**
     * 返回一个延迟动作的触发时间。
     */
    long triggerTime(long delay) {
         //这里一个值得注意的地方是加上了一个now()，
         //另一个是，如果当前delay很大的话，要调用overflowFree来防止溢出。
         return now() +
             ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
    }
    /**
     * Returns nanosecond time offset by origin
     */
    final long now() {
        return System.nanoTime() - NANO_ORIGIN;
    }
    /**
     * 将队列中所有元素的延迟值彼此的和控制在Long.MAX_VALUE以内，避免 
     * 在互相比较时溢出。
     * 这种情况是可能发生的，比如一个满足条件的任务即将出队，这时来了 
     * 一个延迟值是Long.MAX_VALUE的任务。
     */
    private long overflowFree(long delay) {
        Delayed head = (Delayed) super.getQueue().peek();
        if (head != null) {
            long headDelay = head.getDelay(TimeUnit.NANOSECONDS);
            if (headDelay < 0 && (delay - headDelay < 0))
                delay = Long.MAX_VALUE + headDelay;
        }
        return delay;
    }

    //下面这两个方法只是简单的实现，可作为钩子方法由子类实现。

    protected <V> RunnableScheduledFuture<V> decorateTask(
        Runnable runnable, RunnableScheduledFuture<V> task) {
        return task;
    }
    
    protected <V> RunnableScheduledFuture<V> decorateTask(
        Callable<V> callable, RunnableScheduledFuture<V> task) {
        return task;
    }

 

       看下这个ScheduledFutureTask类，首先这个类继承了FutureTask，实现了RunnableScheduledFuture接口。FutureTask之前文章分析过，这里看下后者：

public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {
    /**
     * 是否为周期性任务。
     */
    boolean isPeriodic();
}

 

       看下内部结构：

    private class ScheduledFutureTask<V>
            extends FutureTask<V> implements RunnableScheduledFuture<V> {
        /** 任务序列号 */
        private final long sequenceNumber;
        /** 任务可以执行的时间，单位纳秒 */
        private long time;
        /**
         * 周期性任务的周期时间，单位纳秒。
         * 正数表示固定频率执行，负数表示固定延迟执行，0表示一次性任务。
         */
        private final long period;
        /**
         * Creates a one-shot action with given nanoTime-based trigger time.
         */
        ScheduledFutureTask(Runnable r, V result, long ns) {
            super(r, result);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }
        /**
         * Creates a periodic action with given nano time and period.
         */
        ScheduledFutureTask(Runnable r, V result, long ns, long period) {
            super(r, result);
            this.time = ns;
            this.period = period;
            this.sequenceNumber = sequencer.getAndIncrement();
        }
        /**
         * Creates a one-shot action with given nanoTime-based trigger.
         */
        ScheduledFutureTask(Callable<V> callable, long ns) {
            super(callable);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

 

       由于ScheduledThreadPoolExecutor内部使用延迟队列，而延迟队列中放的元素必须实现Delayed接口，ScheduledFutureTask也必然实现了Delayed接口，看下实现接口方法的细节： 

        public long getDelay(TimeUnit unit) {
            //注意2点：1.延迟值都是按照纳秒时间单位来算的。2.这里减去了now()，还记得之前算触发时间时候加上了now()。
            return unit.convert(time - now(), TimeUnit.NANOSECONDS);
        }
        public int compareTo(Delayed other) {
            if (other == this) // compare zero ONLY if same object
                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;
            }
            //如果要比较的对象不是ScheduledFutureTask，那么按照延迟值进行比较。
            long d = (getDelay(TimeUnit.NANOSECONDS) -
                      other.getDelay(TimeUnit.NANOSECONDS));
            return (d == 0)? 0 : ((d < 0)? -1 : 1);
        }

 

       最后来看一下ScheduledFutureTask类最重要的方法，run方法：

        public void run() {
            //这里首先判断下当前任务是否为周期任务
            if (isPeriodic())
                runPeriodic(); //如果是周期任务，按照周期任务方式运行。
            else
                ScheduledFutureTask.super.run(); //一次性任务的话，就直接执行run方法。
        }

        public boolean isPeriodic() {
            return period != 0;
        }

        private void runPeriodic() {
            //这里执行任务并重置异步任务。
            boolean ok = ScheduledFutureTask.super.runAndReset();
            //判断当前ScheduledThreadPoolExecutor是否关闭。
            boolean down = isShutdown();
            // 如果任务执行成功，
            // 并且ScheduledThreadPoolExecutor没有关闭或者策略允许关闭后继续执行周期任务,
            // 并且ScheduledThreadPoolExecutor没有停止，
            // 那么重新调度任务。
            if (ok && (!down ||
                       (getContinueExistingPeriodicTasksAfterShutdownPolicy() &&
                        !isStopped()))) {
                //重新计算下次触发时间。
                long p = period;
                if (p > 0)
                    time += p; //如果是固定频率，在原有触发时间上加上周期时间。
                else
                    time = triggerTime(-p); //如果是固定延迟，直接指定延迟后的触发时间。
                //算好下次触发时间后，再将任务本身重新加入任务队列。
                ScheduledThreadPoolExecutor.super.getQueue().add(this); 
            }
            // 这可能是最后执行的延迟任务。执行完毕后，
            // 如果当前ScheduledThreadPoolExecutor已关闭，那么中断空闲的工作线程。
            else if (down)
                interruptIdleWorkers();
        }
    public boolean getContinueExistingPeriodicTasksAfterShutdownPolicy() {
        return continueExistingPeriodicTasksAfterShutdown;
    }

 

       看完ScheduledFutureTask类的内容，我们回到schedule方法，看下最后的delayedExecute方法：

    private void delayedExecute(Runnable command) {
        if (isShutdown()) {
            reject(command); //如果当前ScheduledThreadPoolExecutor已关闭，拒绝任务。
            return;
        }
        // 如果当前线程数量小于核心线程数量，那么预启动一个核心线程。
        if (getPoolSize() < getCorePoolSize())
            prestartCoreThread();
        //将任务加入任务队列。
        super.getQueue().add(command);
    }

 

       小总结一下：

              1.ScheduledFutureTask表示可调度的异步任务，提交到ScheduledThreadPoolExecutor的任务都会被包装成这个类。

              2.ScheduledThreadPoolExecutor调度任务时会按照延迟时间来，延迟时间最先到期的任务会被首先调度，如果两个任务延迟时间相同，那么还有内部序列号来保证先入先出的顺序。

              3. ScheduledFutureTask被调度后，具体执行时，会判断自己是否是周期性任务。如果不是，任务执行一次；如果时，先执行任务，执行成功后，会算出下次触发事件(延迟时间)，然后被再次放入ScheduledThreadPoolExecutor的任务队列中，等待下次被调度执行。

 

       了解了ScheduledFutureTask内部执行逻辑，我们再回头看下两个schedule方法，它们内部包装的都是一次性的ScheduledFutureTask，继续看scheduleAtFixedRate和scheduleWithFixedDelay：

    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();
        RunnableScheduledFuture<?> t = decorateTask(command,
            new ScheduledFutureTask<Object>(command,
                                            null,
                                            triggerTime(initialDelay, unit),
                                            unit.toNanos(period)));
        delayedExecute(t);
        return t;
    }
    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();
        RunnableScheduledFuture<?> t = decorateTask(command,
            new ScheduledFutureTask<Boolean>(command,
                                             null,
                                             triggerTime(initialDelay, unit),
                                             unit.toNanos(-delay)));
        delayedExecute(t);
        return t;
    }

       类似于schedule方法，也是先将任务包装成ScheduledFutureTask，但这里构造的是周期性的ScheduledFutureTask；同时可以看到构造固定延迟周期任务时，ScheduledFutureTask构造器传入的第4个参数是负数，这就和上面分析的ScheduledFutureTask内部的period定义对上了：负数表示固定延迟周期性任务。

 

• 最后再看一下和ScheduledThreadPoolExecutor关闭相关的一些特殊处理：

    public void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value) {
        continueExistingPeriodicTasksAfterShutdown = value;
        if (!value && isShutdown())
            cancelUnwantedTasks();
    }
    public void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value) {
        executeExistingDelayedTasksAfterShutdown = value;
        if (!value && isShutdown())
            cancelUnwantedTasks();
    }
    public void shutdown() {
        cancelUnwantedTasks();
        super.shutdown();
    }
    private void cancelUnwantedTasks() {
        //关闭后是否继续执行延迟的任务。
        boolean keepDelayed = getExecuteExistingDelayedTasksAfterShutdownPolicy();
        //关闭后是否继续执行周期性的任务。
        boolean keepPeriodic = getContinueExistingPeriodicTasksAfterShutdownPolicy();
        if (!keepDelayed && !keepPeriodic)
            super.getQueue().clear(); //如果都不继续，那么直接清空任务队列。
        else if (keepDelayed || keepPeriodic) {
            //否则会按照相应的策略来取消相应类型的任务。
            Object[] entries = super.getQueue().toArray();
            for (int i = 0; i < entries.length; ++i) {
                Object e = entries[i];
                if (e instanceof RunnableScheduledFuture) {
                    RunnableScheduledFuture<?> t = (RunnableScheduledFuture<?>)e;
                    if (t.isPeriodic()? !keepPeriodic : !keepDelayed)
                        t.cancel(false);
                }
            }
            entries = null;
            //最后清理一把任务队列里面被取消的任务。
            purge();
        }
    }

       可以看到在设置关闭后任务处理策略和关闭当前ScheduledThreadPoolExecutor时，都会(按需)调用一下cancelUnwantedTasks方法来清理不需要的任务。

 

       ScheduledThreadPoolExecutor的代码解析完毕！ 

 

 

       参见：Jdk1.6 JUC源码解析(17)-ThreadPoolExecutor

                  Jdk1.6 JUC源码解析(18)-DelayQueue