Java并发--ScheduledThreadPoolExecutor

ゝ一世哀愁。 2021-08-14 01:21 418阅读 0赞

### ScheduledThreadPoolExecutor ###

**ScheduledThreadPoolExecutor继承自ThreadPoolExecutor**。它主要用来在给定的延迟之后运行任务，或者定期执行任务。ScheduledThreadPoolExecutor的功能与Timer类似，但ScheduledThreadPoolExecutor功能更强大、更灵活。

**Timer对应的是单个后台线程，而ScheduledThreadPoolExecutor可以在构造函数中指定多个对应的后台线程数。**

### ScheduledThreadPoolExecutor的运行机制 ###

ScheduledThreadPoolExecutor的执行示意图（本文基于JDK 6）如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

**DelayQueue是一个无界队列**，所以ThreadPoolExecutor的maximumPoolSizeScheduledThreadPoolExecutor中没有什么意义（设置maximumPoolSize的大小没有什么效果）。

ScheduledThreadPoolExecutor的执行主要分为两大部分：

1.  当调用ScheduledThreadPoolExecutor的scheduleAtFixedRate()方法或scheduleWithFixedDelay()方法时，会向ScheduledThreadPoolExecutor的DelayQueue添加一个实现了RunnableScheduledFutur接口的ScheduledFutureTask。
2.  线程池中的线程从DelayQueue中获取ScheduledFutureTask，然后执行任务。

ScheduledThreadPoolExecutor的任务传递示意图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

ScheduledThreadPoolExecutor为了实现周期性的执行任务，对ThreadPoolExecutor做了如下  
的修改：

*  使用DelayQueue作为任务队列。
 *  获取任务的方式不同（后文会说明）。
 *  执行周期任务后，增加了额外的处理（后文会说明）。

### ScheduledThreadPoolExecutor的实现 ###

ScheduledThreadPoolExecutor会把待调度的任务（ScheduledFutureTask）放到一个DelayQueue中。

ScheduledFutureTask主要包含3个成员变量，如下：

*  long型成员变量time，表示这个任务将要被执行的具体时间。
 *  long型成员变量sequenceNumber，表示这个任务被添加到ScheduledThreadPoolExecutor中的序号。
 *  long型成员变量period，表示任务执行的间隔周期。

DelayQueue封装了一个PriorityQueue，这个PriorityQueue会对队列中的ScheduledFutureTask进排序。排序时，time小的排在前面（时间早的任务将被先执行）。如果两个ScheduledFutureTask的time相同，就比较sequenceNumber，sequenceNumber小的排在前面（也就是说，如果两个任务的执行时间相同，那么先提交的任务将被先执行）。

ScheduledThreadPoolExecutor中的线程1执行某个周期任务的4个步骤：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]

1.  线程1从DelayQueue中获取已到期的ScheduledFutureTask（DelayQueue.take()）。到期任务是指ScheduledFutureTask的time大于等于当前时间。
2.  线程1执行这个ScheduledFutureTask。
3.  线程1修改ScheduledFutureTask的time变量为下次将要被执行的时间。
4.  线程1把这个修改time之后的ScheduledFutureTask放回DelayQueue中（DelayQueue.add()）。

下面是DelayQueue.take()方法的源代码实现：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]

DelayQueue.take()的执行示意图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]

如图所示，获取任务分为3大步骤：

1.  **获取Lock**。
2.  **获取周期任务**。
    
     *  如果PriorityQueue为空，当前线程到Condition中等待；否则执行下面的2.2。
     *  如果PriorityQueue的头元素的time时间比当前时间大，到Condition中等待到time时间；否则执行下面的2.3。
     *  获取PriorityQueue的头元素（2.3.1）；如果PriorityQueue不为空，则唤醒在Condition中等待的所有线程（2.3.2）。
3.  **释放Lock**。  
    ScheduledThreadPoolExecutor在一个循环中执行步骤2，直到线程从PriorityQueue获取到一  
    个元素之后（执行2.3.1之后），才会退出无限循环（结束步骤2）。

ScheduledThreadPoolExecutor中的线程执行任务的步骤4，把ScheduledFutureTask放入DelayQueue中的过程。下面是DelayQueue.add()的源代码实现。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]

DelayQueue.add()的执行示意图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]如图所示，添加任务分为3大步骤。

1.  **获取Lock**。
2.  **添加任务**。
    
     *  向PriorityQueue添加任务。
     *  如果在上面2.1中添加的任务是PriorityQueue的头元素，唤醒在Condition中等待的所有线程。
3.  **释放Lock**。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/2781c5db375a4b20b476295bea8e6dba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/05c397ad43e44a2a93292f59a8d47053.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/73802b1ec57143d89e8ebfb3ebd77c46.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/30530bfb7e064f06b03f5f06fa8f10af.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/8dce1e05cc034b30a5854c647c56465e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/a15466add14347368265f309d6567b50.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/d3dc0578833c49ac851b77e64958b064.png