并发包中ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor继承了ThreadPoolExecutor,是一个定时任务调度执行的线程池

一、变量与构造函数

    /**
     * shutdown时是否取消定时任务
     */
    private volatile boolean continueExistingPeriodicTasksAfterShutdown;

    /**
     * shutdown时是否取消定时任务
     */
    private volatile boolean executeExistingDelayedTasksAfterShutdown = true;

    /**
     * ScheduledFutureTask.cancel 任务取消是否删除队列中的任务
     */
    private volatile boolean removeOnCancel = false;

    /**
     * Sequence number to break scheduling ties, and in turn to
     * guarantee FIFO order among tied entries.
     */
    private static final AtomicLong sequencer = new AtomicLong();

    //构造方法默认：
    //maximumPoolSize == Integer.MAX_VALUR,
    //KeepAliveTime == 0,
    //workQueue == new DelayedWorkQueue(),
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

    /**
     * Creates a new {@code ScheduledThreadPoolExecutor} with the
     */
    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue(), threadFactory);
    }

    /**
     * Creates a new ScheduledThreadPoolExecutor with the given
     */
    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue(), handler);
    }

    /**
     * Creates a new ScheduledThreadPoolExecutor with the given
     */
    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory,
                                       RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue(), threadFactory, handler);
    }

ScheduledThreadPoolExecutor采用自定义的内部类DelayedWorkQueue的实例作为工作队列，DelayedWorkQueue类似于DelayQueue

二、重要实现

ScheduledThreadPoolExecutor自定义了两个内部类DelayedWorkQueue、ScheduledFutureTask，现在看看ScheduledFutureTask构造

1.ScheduledFutureTask类

    private class ScheduledFutureTask<V>
            extends FutureTask<V> implements RunnableScheduledFuture<V> {//继承FutureTask

        /** 自增序列号，任务执行时间相同时，序列号小的先执行（FIFO），由上面的AtomicLong原子变量sequence保证线程安全*/
        private final long sequenceNumber;

        /** 任务执行的时间yyyyMMddHHmmss，单位纳秒 */
        private long time;

        /**
         * 任务周期，单位是纳秒
         * period == 0 时，当前任务为一次性任务，执行完后退出
         * period > 0 时，当前任务为fixed-rate任务，是固定频率的定时任务，具体一点：定时7点执行第一次任务，之后每小时执行一次任务：7/8/9...
         * period < 0 时，当前任务为fixed-delay任务，是固定延时的定时任务，具体一点：定时7点执行第一次任务，第一次任务执行完成的时间+1小时...即可能是7/8.1/9.2...
         */
        private final long period;

        /** The actual task to be re-enqueued by reExecutePeriodic */
        RunnableScheduledFuture<V> outerTask = this;

        /**
         * Index into delay queue, to support faster cancellation.
         */
        int heapIndex;

        /**
         * 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 time.
         */
        ScheduledFutureTask(Callable<V> callable, long ns) {
            super(callable);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        public long getDelay(TimeUnit unit) {
            return unit.convert(time - now(), NANOSECONDS);
        }

        //延迟队列比较器
        //同一任务返回0
        //不同任务先比较下一次执行的时间，若相同则比较任务初始化时的序列号sequenceNumber
        public int compareTo(Delayed other) {
            if (other == this) // compare zero 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;
            }
            long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
            return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
        }

        /**
         * 是周期重复执行任务
         */
        public boolean isPeriodic() {
            return period != 0;
        }

        /**
         * 设置周期任务的下一次执行的时间
         */
        private void setNextRunTime() {
            long p = period;
            if (p > 0)
                time += p;
            else
                time = triggerTime(-p);
        }

        public boolean cancel(boolean mayInterruptIfRunning) {
            boolean cancelled = super.cancel(mayInterruptIfRunning);
            if (cancelled && removeOnCancel && heapIndex >= 0)
                remove(this);
            return cancelled;
        }

        /**
         * 定时处理方法，ThreadPoolExecutor.execute(command)中command.run(）== ScheduledFutureTask.super.run()，给它再包装了一次run方法
         * 一次性任务直接执行
         * 定时任务①执行②设置下一次执行时间③入DelayedWorkerQueue
         */
        public void run() {
            boolean periodic = isPeriodic();
            if (!canRunInCurrentRunState(periodic))
                cancel(false);
            else if (!periodic)
                ScheduledFutureTask.super.run();
            else if (ScheduledFutureTask.super.runAndReset()) {
                setNextRunTime();
                reExecutePeriodic(outerTask);
            }
        }
    

2.setNextRunTime（）:返回任务的下一次执行时间

    private void setNextRunTime() {
        long p = period;
        if (p > 0)
            time += p;//固定频率定时任务  time = firstTime + n*p
        else
            time = triggerTime(-p);//固定延时定时任务  time = firstTime;第二次任务执行时间是firstTask执行完后+延时（now + delay）
    }

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

3.ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit):创建一次性任务(period == 0)

    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        //创建一次性定时任务，period == 0，指定下次执行的时间time
        RunnableScheduledFuture<?> t = decorateTask(command,
            new ScheduledFutureTask<Void>(command, null,
                                          triggerTime(delay, unit)));
        //任务入延迟队列DelayedWorkerQueue
        delayedExecute(t);
        return t;
    }

    private void delayedExecute(RunnableScheduledFuture<?> task) {
        if (isShutdown())
            //线程池状态检验
            reject(task);
        else {
            //任务入延迟工作队列
            super.getQueue().add(task);
            if (isShutdown() &&
                !canRunInCurrentRunState(task.isPeriodic()) &&
                remove(task))
                //线程池正在关闭，
                //
                //延迟工作队列删除任务，并尝试终止线程池
                task.cancel(false);
            else
                //线程池RUNNING状态，工作线程数<corePoolSize，创建工作线程add(null，boolean)
                ensurePrestart();
        }
    }

4.ScheduledFuture<?>  scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit):创建固定延时的定时任务(period < 0)

    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();
        //创建固定延时的定时任务period <0,并设置下次运行的时间
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(-delay));
        //设置outerTask
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        //任务执行（入延迟工作队列）
        delayedExecute(t);
        return t;
    }

5.ScheduledFuture<?>  scheduleWithFixedRate(Runnable command, long initialDelay, long delay, TimeUnit unit):创建固定频率的定时任务

    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();
        //创建固定频率的定时任务period > 0,并设置下次运行时间
        ScheduledFutureTask<Void> sft =
            new ScheduledFutureTask<Void>(command,
                                          null,
                                          triggerTime(initialDelay, unit),
                                          unit.toNanos(period));
        //设置outerTask
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        //执行定时任务（入延迟工作队列）
        delayedExecute(t);
        return t;
    }

 三.总结

1.是定时任务线程池，继承ThreadPoolExcutor实现，自定义了延迟队列DelayedWorkerQueue,对command在封装成ScheduledFutureTask

2.用过ScheduledFutureTask的period参数来区分为3种任务

①一次性任务（period == 0）

②固定频率定时任务（fixed-rate : period > 0）7/8/9 上一次任务开始时间+period

③固定延时定时任务（fixed-delay : period < 0）7/8.1/9.2 上一次任务结束时间+period

四.实例

public class SchedulePoolTest {

    private static final ScheduledExecutorService pool = Executors.newScheduledThreadPool(2);
    private static final Lock lock = new ReentrantLock();

    public static String format(long time){
        Date date = new Date(time);
        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(date);
    }

    public static void main(String[] args){
        Runnable task = new Runnable() {
            @Override
            public void run() {
                lock.lock();
                try {
                    System.out.println(format(System.currentTimeMillis()));
                }finally {
                    lock.unlock();
                }
            }
        };
        System.out.println(format(System.currentTimeMillis()));
        //pool.schedule(task,5,TimeUnit.SECONDS);
        //pool.scheduleAtFixedRate(task,5,1,TimeUnit.SECONDS);
        Runnable task1 = new Runnable() {
            @Override
            public void run() {
                lock.lock();
                try {
                    System.out.println(format(System.currentTimeMillis()));
                    Thread.sleep(2000);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }finally {
                    lock.unlock();
                }
            }
        };
        pool.scheduleWithFixedDelay(task1,5,1,TimeUnit.SECONDS);
        try {
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        pool.shutdown();

    }

}

 参考自《java并发编程之美》