并发之定时任务线程池

╰+哭是因爲堅強的太久メ 2022-10-18 00:52 160阅读 0赞

## 1.ScheduledThreadPoolExecutor 作用 ##

*  指定延时后执行任务
 *  周期性重复执行任务

## 2. ScheduledThreadPoolExecutor 类结构图 ##

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

## 3. ScheduledThreadPoolExecutor的运行机制 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZWd1Y2hlbg_size_16_color_FFFFFF_t_70 1]  
**注意：** DelayQueue是无界队列（maximumPoolSize参数无效）；DelayQueue内部封装了一个PriorityQueue，它会根据time的先后时间排序，若time相同则根据sequenceNumber排序

**流程：**

1.  当调用ScheduledThreadPoolExecutor的scheduleAtFixedRate()方法或者scheduleWithFixedDelay()方法时 —> 向ScheduledThreadPoolExecutor的DelayQueue任务队列添加一个ScheduledFutureTask（ScheduledFutureTask实现RunnableScheduledFutur接口）
2.  线程池中的线程从DelayQueue中获取已经到期ScheduledFutureTask，然后执行任务
3.  执行结束后重新设置任务的到期时间，再次放回DelayQueue

## 4.SchduledFutureTask ##

#### 4.1 成员变量 ####

1.  private long time：任务开始的时间
2.  private final long sequenceNumber：任务的序号
3.  private final long period：任务执行的时间间隔  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZWd1Y2hlbg_size_16_color_FFFFFF_t_70 2]

#### 4.2 Run方法 ####

1.  如果当前线程池已经不支持执行任务，则取消该任务并返回，否则执行下一步
2.  如果不是周期性任务，调用FutureTask中的run方法执行，会设置执行结果，然后直接返回，否则执行下一步
3.  如果是周期性任务，调用FutureTask中的runAndReset方法执行，不会设置执行结果，然后直接返回，否则执行第四步和第五步
4.  计算下次执行该任务的具体时间
5.  重复执行任务

/** * Overrides FutureTask version so as to reset/requeue if periodic. */
            public void run() { 
                boolean periodic = isPeriodic();
                // 如果当前线程池已经不支持执行任务，则取消该任务
                if (!canRunInCurrentRunState(periodic))
                    cancel(false);
                // 如果不需要周期性执行，则直接执行run方法然后结束
                else if (!periodic)
                    ScheduledFutureTask.super.run();
                // 如果需要周期执行，则在执行完任务以后，设置下一次执行时间
                else if (ScheduledFutureTask.super.runAndReset()) { 
                    // 计算下次执行该任务的时间
                    setNextRunTime();
                    // 重复执行任务
                    reExecutePeriodic(outerTask);
                }
            }

## 5. 线程池任务的提交 ##

**schedule方法：** 任务在指定延迟时间到达后触发，只会执行一次

/** * @throws RejectedExecutionException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */
        public ScheduledFuture<?> schedule(Runnable command,
                                           long delay,
                                           TimeUnit unit) { 
            if (command == null || unit == null)
                throw new NullPointerException();
            // 嵌套结构，首先把用户提交的任务包装成ScheduledFutureTask.然后在调用decorateTask进行包装
            RunnableScheduledFuture<?> t = decorateTask(command,
                new ScheduledFutureTask<Void>(command, null,
                                              triggerTime(delay, unit)));
            // 包装好任务以后，就进行提交了
            delayedExecute(t);
            return t;
        }
    
    
        /** * Main execution method for delayed or periodic tasks. If pool * is shut down, rejects the task. Otherwise adds task to queue * and starts a thread, if necessary, to run it. (We cannot * prestart the thread to run the task because the task (probably) * shouldn't be run yet.) If the pool is shut down while the task * is being added, cancel and remove it if required by state and * run-after-shutdown parameters. * * @param task the task */
        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
                    // 增加一个worker线程，避免提交的任务没有worker去执行(因为该类没有像ThreadPoolExecutor一样，woker满了才放入队列)
                    ensurePrestart();
            }
        }

## 6. DelayedWorkQueue ##

DelayedWorkQueue是一个基于堆的数据结构，类似于DelayQueue和PriorityQueue。

**作用：** **按照执行时间的升序来排列**，执行时间距离当前时间越近的任务在队列的前面（注意：这里的顺序并不是绝对的，堆中的排序只保证了子节点的下次执行时间要比父节点的下次执行时间要大，而叶子节点之间并不一定是顺序的）

**示意图：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZWd1Y2hlbg_size_16_color_FFFFFF_t_70 3]  
**应用场景：**

定时任务执行时需要取出最近要执行的任务，所以任务在队列中每次出队时一定要是当前队列中执行时间最靠前的

DelayedWorkQueue是一个优先级队列，它可以**保证每次出队的任务都是当前队列中执行时间最靠前的**，由于它是基于堆结构的队列，堆结构在执行插入和删除操作时的最坏时间复杂度是 O(logN)。

#### 6.1 DelayedWorkQueue属性 ####

// 队列初始容量
            private static final int INITIAL_CAPACITY = 16;
            // 根据初始容量创建RunnableScheduledFuture类型的数组
            private RunnableScheduledFuture<?>[] queue =
                new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
            private final ReentrantLock lock = new ReentrantLock();
            private int size = 0;
    
            // leader线程
            private Thread leader = null;
    
            // 当较新的任务在队列的头部可用时，或者新线程可能需要成为leader，则通过该条件发出信号
            private final Condition available = lock.newCondition();

**注意：**

多线程网络模型：

线程会有三种身份中的一种：**leader**和**follower**，以及一个干活中的状态：**proccesser**

基本原则：永远最多只有一个leader，而所有follower都在等待成为leader。

线程池启动时会自动产生一个Leader负责等待网络IO事件，当有一个事件产生时，Leader线程首先通知一个Follower线程将其提拔为新的Leader，然后自己执行这个任务，去处理这个网络事件，处理完毕后加入Follower线程等待队列，等待下次成为Leader。

这种方法可以增强CPU高速缓存相似性，及消除动态内存分配和线程间的数据交换。

## 7. ScheduledThreadPoolExecutor与Timer的区别 ##

1.  Timer是单线程模式，ScheduledThreadPoolExecutor是多线程模式，并且重用线程池
2.  Timer调度是基于操作系统的绝对时间的，ScheduledThreadPoolExecutor调度是基于相对时间的，不受操作系统时间改变的影响
3.  Timer不会捕获TimerTask抛出的异常，加上Timer又是单线程的。一旦某个调度任务出现异常，则整个线程就会终止，其他需要调度的任务也不再执行。ScheduledThreadPoolExecutor基于线程池来实现调度功能，某个任务抛出异常后，其他任务仍能正常执行。
4.  Timer中执行的TimerTask任务整体上没有优先级的概念，只是按照系统的绝对时间来执行任务。ScheduledThreadPoolExecutor距离下次执行的时间间隔短的任务的优先级比较高
5.  Timer不支持对任务的排序。ScheduledThreadPoolExecutor为需要调度的每个任务按照距离下次执行时间间隔的大小来排序
6.  Timer无法从TimerTask中获取返回的结果。ScheduledThreadPoolExecutor中FutureTask类，能够通过Future来获取返回的结果。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZWd1Y2hlbg_size_16_color_FFFFFF_t_70]: /images/20221014/8023cdeed99d4e669bca762092b157f0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZWd1Y2hlbg_size_16_color_FFFFFF_t_70 1]: /images/20221014/4e5011a09eb54a0ebc86dfe8f48340d5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZWd1Y2hlbg_size_16_color_FFFFFF_t_70 2]: /images/20221014/3f08919db8ca44b98211b65ac06bb8d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZWd1Y2hlbg_size_16_color_FFFFFF_t_70 3]: /images/20221014/2d78a4f645604fbf8edcfcc31d2acf3c.png