JUC之ScheduledThreadPoolExecutor

╰+哭是因爲堅強的太久メ 2022-09-01 00:54 187阅读 0赞

在《[JUC之Executors][JUC_Executors]》中提到，可以通过**Executors**提供的方法**newScheduledThreadPool**创建一种**ScheduledThreadPoolExecutor**类型的线程池。

ScheduledThreadPoolExecutor，一种特殊的**ThreadPoolExecutor**。Scheduled意为预定的，即该线程池可以预定在某个时间开始执行（延迟执行）或者周期性间隔执行（周期性循环执行）。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70][]

ScheduledThreadPoolExecutor继承了ThreadPoolExecutor，因此其具有基本线程池的相关功能。其还实现了**ScheduledExecutorService**接口对线程池的功能进行了扩展。在接口ScheduledExecutorService中定义了与定时执行功能的相关方法。

--------------------

# **█ **ScheduledExecutorService #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1][]

### 方法： ###

ScheduledExecutorService继承了**ExecutorService**接口。*（关于ExecutorService接口的介绍，请戳《[JUC之ThreadPoolExecutor（1）总体介绍][JUC_ThreadPoolExecutor_1]》）*

*  **schedule**

public ScheduledFuture<?> schedule(Runnable command,
                                       long delay, TimeUnit unit);

public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay, TimeUnit unit);

schedule有两个重载方法，区别在于一个支持参数为**Runnable**类型的任务，一个支持参数为**Callable**类型的任务。另外两个参数分别**delay**设置延迟执行的时长以及**unit**设置delay时长的单位（毫秒、秒、小时等）。该方法的作用是执行该方法提交的任务，在等待delay个时间单位后才开始运行任务，任务从提交到运行要等待delay个时间单位。但任务只会被执行一次。（只具有延迟执行，不具有周期性循环执行的功能）

方法返回值类型为**ScheduledFuture**：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2][]

*  **scheduleAtFixedRate**

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);

具有延迟执行以及周期性循环执行的功能。**initialDelay**设置延迟执行的时长，period设置两次任务成功执行之间的间隔时间（周期性循环执行中的周期间隔时长），即前一次任务成功执行后间隔period个时间长度再去执行下一次任务。unit设置initialDelay与period的时间单位。

*  **scheduleWithFixedDelay**

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);

与scheduleAtFixedRate方法相同，既具有延迟执行也具有周期性循环执行的功能。每个参数的含义相同，区别在于这里的delay时长，是两个任务终止之间的间隔，即前一次任务终止后间隔delay个时间长度再去执行下一次任务。*（查看源码时再比较scheduleAtFixedRate与scheduleWithFixedDelay的具体区别）*

--------------------

# █ DelayedWorkQueue #

线程池需要一个阻塞队列（**BlockingQueue**），ScheduledThreadPoolExecutor作为线程池的一种扩展，自然也需要一个阻塞队列。在ScheduledThreadPoolExecutor中使用的阻塞队列为**DelayedWorkQueue**。*（关于BlockingQueue阻塞队列的介绍，请戳《[BlockingQueue初识][BlockingQueue]》）*

DelayedWorkQueue是在ScheduledThreadPoolExecutor中定义的一个内部类。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 4][]

关于DelayedWorkQueue的详细介绍，请戳《[DelayedWorkQueue与ScheduledFutureTask][DelayedWorkQueue_ScheduledFutureTask]》

--------------------

# █ ScheduledThreadPoolExecutor #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 5][]

ScheduledThreadPoolExecutor继承了ThreadPoolExecutor，因此其具有ThreadPoolExecutor的相关功能，关于ThreadPoolExecutor部分的功能，在这里就不介绍了，感兴趣的同学请戳《[JUC之ThreadPoolExecutor（1）总体介绍][JUC_ThreadPoolExecutor_1]》、《[JUC之ThreadPoolExecutor（2）方法介绍][JUC_ThreadPoolExecutor_2]》。我们只介绍ScheduledThreadPoolExecutor实现了接口ScheduledExecutorService中的方法，即ScheduledThreadPoolExecutor的延迟执行以及周期性循环执行的功能。

### 构造器： ###

①自定义核心**worker**数量，最大worker数量为**Integer.MAX\_VALUE**。

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

②不仅可以指定核心worker数量，还可以指定线程工厂。

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

③指定核心worker数量和拒绝策略

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

④指定核心worker数量、线程工厂、拒绝策略

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

ScheduledThreadPoolExecutor设置的阻塞队列都为DelayedWorkQueue。

super()，调用父类的构造器，即调用ThreadPoolExecutor的构造方法。

### 方法： ###

*  **schedule**

前面介绍过在接口ScheduledExecutorService中定义了两个重载的schedule方法，分别支持参数类型为Runnable的任务和类型为Callable的任务。

参数类型为Runnable：

public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        RunnableScheduledFuture<?> t = decorateTask(command,
            new ScheduledFutureTask<Void>(command, null,
                                          triggerTime(delay, unit)));
        delayedExecute(t);
        return t;
    }

参数类型为Callable：

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

不管是类型是Runnable还是Callable，都要封装到**ScheduledFutureTask**对象中，然后再封装成**RunnableScheduledFuture**对象。

new ScheduledFutureTask：

创建ScheduledFutureTask对象，用于ScheduledThreadPoolExecutor线程池执行的任务。

ScheduledFutureTask(Runnable r, V result, long ns) {
        super(r, result);
        this.time = ns;
        this.period = 0;
        this.sequenceNumber = sequencer.getAndIncrement();
    }

**decorateTask**：

在ScheduledThreadPoolExecutor中decorateTask方法并没有做任何事情，仅仅是将参数传进来的RunnableScheduledFuture对象再返回出去。子类可通过重写此方法完成自定义的功能逻辑。

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

**delayedExecute：**

private void delayedExecute(RunnableScheduledFuture<?> task) {
        // 如果线程池已经关闭，则拒绝该任务。
        if (isShutdown())
            reject(task);
        else {
            // 将任务加到等待队列中，即DelayedWorkQueue中
            super.getQueue().add(task);
            if (isShutdown() &&
                !canRunInCurrentRunState(task.isPeriodic()) &&
                remove(task))
                task.cancel(false);
            else
                // 执行任务
                ensurePrestart();
        }
    }

**reject：**

拒绝任务就是使用拒绝策略拒绝当前提交的任务，该方法是父类ThreadPoolExecutor中提供的方法。

final void reject(Runnable command) {
        handler.rejectedExecution(command, this);
    }

**ensurePrestart**：

该方法也是父类ThreadPoolExecutor中提供的方法。

void ensurePrestart() {
        int wc = workerCountOf(ctl.get());
        if (wc < corePoolSize)
            addWorker(null, true);
        else if (wc == 0)
            addWorker(null, false);
    }

*  **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));
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

new ScheduledFutureTask：

ScheduledFutureTask(Runnable r, V result, long ns, long period) {
        super(r, result);
        this.time = ns;
        this.period = period;
        this.sequenceNumber = sequencer.getAndIncrement();
    }

scheduleAtFixedRate与schedule分别在创建ScheduledFutureTask对象时，scheduleAtFixedRate方法可以指定period周期值，而schedule无法指定period周期值。因此scheduleAtFixedRate具有周期性定时执行功能，而schedule不具有。

*  **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));
        RunnableScheduledFuture<Void> t = decorateTask(command, sft);
        sft.outerTask = t;
        delayedExecute(t);
        return t;
    }

scheduleWithFixedDelay方法与scheduleAtFixedRate的区别在于创建ScheduledFutureTask对象关于period值的处理不同。scheduleWithFixedDelay中是unit.toNanos(-delay)，对参数间隔时间的值进行取反，又因为delay只能要求是正数，所以-delay是负数。而scheduleAtFixedRate中是unit.toNanos(period)，直接使用的是参数间隔时间的值。

--------------------

#  █ 写在最后 #

ScheduledThreadPoolExecutor的功能实现主要就是将Runnable或Callable类型的任务封装成ScheduledFutureTask，然后将ScheduledFutureTask任务添加到DelayedWorkQueue队列中。任务的调度和执行仍然还是依赖于ThreadPoolExecutor的功能实现，调用ThreadPoolExecutor提供的ensurePrestart方法启动线程池开始调度任务。

因此ScheduledThreadPoolExecutor的功能实现关键还是在于DelayedWorkQueue与ScheduledFutureTask的实现。在ScheduledThreadPoolExecutor中，是将提交的任务封装成ScheduledFutureTask对象，ScheduledFutureTask也实现了Runnable接口，执行任务就是执行ScheduledFutureTask的run方法或者将ScheduledFutureTask对象添加到DelayedWorkQueue阻塞队列中，任务调度和执行交给ThreadPoolExecutor的实现去完成。

**ScheduledFutureTask\#run**

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

①isPeriodic：

当前任务是否为周期性定时执行任务，即period==0就不是周期性定时执行任务，period != 0就是。前面看见的schedule方法中在创建ScheduledFutureTask对象时，指定的period为0，因此schedule方法执行的就不是周期性定时任务。而scheduleAtFixedRate与scheduleWithFixedDelay方法中period限制为大于0的数（正数），因此这两个方法执行的就是周期性定时任务。

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

②if (!periodic） ScheduledFutureTask.super.run();

如果不是周期性定时任务，就只调用run方法执行一次。

③ScheduledFutureTask.super.runAndReset()

**runAndReset**方法是**FutureTask**中提供的一个方法，使用CAS乐观锁来保证线程安全。从其中的逻辑可以看出，只有ScheduledFutureTask的状态为NEW状态的时候才会执行。

protected boolean runAndReset() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return false;
        boolean ran = false;
        int s = state;
        try {
            Callable<V> c = callable;
            if (c != null && s == NEW) {
                try {
                    c.call(); // don't set result
                    ran = true;
                } catch (Throwable ex) {
                    setException(ex);
                }
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
        return ran && s == NEW;
    }

④setNextRunTime

设置下一次该任务被执行的时间。time是在创建ScheduledFutureTask任务时设置的间隔时长，period是周期间隔时长。

private void setNextRunTime() {
        long p = period;
        if (p > 0)
            time += p;
        else
            time = triggerTime(-p);
    }

⑤reExecutePeriodic(outerTask);

注意这个方法的参数是outerTask，而不是this。outerTask则是在scheduleAtFixedRate、scheduleWithFixedDelay方法中通过sft.outerTask = t;指定，本质上也就是reExecutePeriodic(this)。

void reExecutePeriodic(RunnableScheduledFuture<?> task) {
        if (canRunInCurrentRunState(true)) {
            super.getQueue().add(task);
            if (!canRunInCurrentRunState(true) && remove(task))
                task.cancel(false);
            else
                ensurePrestart();
        }
    }

[JUC_Executors]: https://blog.csdn.net/weixin_50518271/article/details/118768318?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70]: /images/20220829/efbdb842b8784852b6ecc1b47057469c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1]: /images/20220829/7446f2c957714f4d9983104a0b28ec95.png
[JUC_ThreadPoolExecutor_1]: https://blog.csdn.net/weixin_50518271/article/details/118684239?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2]: /images/20220829/c6b742b6dc6f468d8c6aa158c088febc.png
[BlockingQueue]: https://blog.csdn.net/weixin_50518271/article/details/117789512?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 3]: /images/20220829/adc27a9352f7479faa2ba7c1b7e370ea.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 4]: /images/20220829/8af588ee03424660a9037bf16746b148.png
[DelayedWorkQueue_ScheduledFutureTask]: https://blog.csdn.net/weixin_50518271/article/details/119087739?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 5]: /images/20220829/9ada1cf92c0e49cd8680e667e7262bec.png
[JUC_ThreadPoolExecutor_2]: https://blog.csdn.net/weixin_50518271/article/details/118735011?spm=1001.2014.3001.5501