六. ScheduledThreadPoolExecutor原理解析-任务提交

男娘i 2024-03-29 14:42 99阅读 0赞

### 前言 ###

`ScheduledThreadPoolExecutor`继承于`ThreadPoolExecutor`，扩展实现了**延时执行任务**和**定时执行任务**的功能。`ScheduledThreadPoolExecutor`存储任务的队列为`DelayedWorkQueue`，是一个基于**小根堆**实现的延时优先级队列，`ScheduledThreadPoolExecutor`会将每一个提交到线程池的任务先封装为`ScheduledFutureTask`，然后再插入到`DelayedWorkQueue`中。本篇文章将结合源码，先对`ScheduledThreadPoolExecutor`的任务提交进行分析。

### 正文 ###

`ScheduledThreadPoolExecutor`提供了四个提交任务的方法，如下所示。

// 提交无返回值的延时任务
    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;
    }
    
    // 提交有返回值的延时任务
    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;
    }
    
    // 提交固定周期的定时任务
    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;
    }
    
    // 提交固定延时的定时任务
    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;
    }

可知`ScheduledThreadPoolExecutor`支持执行如下的任务。

*  **无返回值的延时任务**；
 *  **有返回值的延时任务**；
 *  **固定周期的定时任务**；
 *  **固定延时的定时任务**。

这里解释一下**固定周期的定时任务**和**固定延时的定时任务**。**固定周期的定时任务**的下一次执行时间点为上一次执行时间点加上周期时间，而**固定延时的定时任务**的下一次执行时间点为上一次执行结束时间点加上延时时间。

在提交任务的四个方法中，均是将提交的任务封装为`ScheduledFutureTask`，在将任务封装成`ScheduledFutureTask`时需要指定任务首次执行的时间点（即初始延时），和任务的执行间隔（为正值表示固定周期，为负值表示固定时延，为0表示仅执行一次），而指定任务首次执行的时间点时，为了防止时间点的值不在长整型的最大值范围内，需要在`triggerTime()`方法中进行处理，如下所示。

// 获取延时操作的触发时间
    // 触发时间 = 当前时间 + 延时时间
    private long triggerTime(long delay, TimeUnit unit) {
        
    	return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
    }
    
    // 如果延时时间大于等于长整型最大值的一半，则执行overflowFree()方法对延时时间进行处理
    // 执行overflowFree()方法就是为了使得最终得到的触发时间的值在长整型最大值以内，以防止compareTo()时值溢出
    long triggerTime(long delay) {
        
    	return now() +
    		((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
    }

封装好的`ScheduledFutureTask`会在`delayedExecute()`方法中添加到延时队列，如下所示。

private void delayedExecute(RunnableScheduledFuture<?> task) {
        
    	// 如果线程池状态不是RUNNING，则执行拒绝策略
    	if (isShutdown())
    		reject(task);
    	else {
        
    		// 将任务添加到延时队列中
    		super.getQueue().add(task);
    		// 此时如果线程池状态为非RUNNING，并且线程池策略为非RUNNING状态下延时任务或定时任务不再执行，则将任务从延时队列中删除
    		// 将任务从延时队列中删除后，还需要关闭任务，如果任务尚未执行，那么关闭任务后任务就不会再被执行
    		// 如果任务正在被执行，则不会尝试中断执行任务的Worker来关闭任务
    		if (isShutdown() &&
    			!canRunInCurrentRunState(task.isPeriodic()) &&
    			remove(task))
    			task.cancel(false);
    		else
    			// 创建初始任务为null的Worker，即使核心线程数为0也需要确保线程池至少有一个Worker在工作
    			ensurePrestart();
    	}
    }
    
    void ensurePrestart() {
        
    	int wc = workerCountOf(ctl.get());
    	if (wc < corePoolSize)
    		addWorker(null, true);
    	else if (wc == 0)
    		addWorker(null, false);
    }

如果线程池状态为**RUNNING**，那么会将封装好的`ScheduledFutureTask`添加到延时队列，同时会创建一个初始任务为**null**的`Worker`，之所以创建的`Worker`初始任务为**null**，是因为`ScheduledThreadPoolExecutor`中的任务`ScheduledFutureTask`都会有一个初始延时，即提交到线程池的任务不会立即被执行，所以`Worker`均需要到延时队列中去获取任务。

### 总结 ###

`ScheduledThreadPoolExecutor`支持执行如下的任务。

*  **无返回值的延时任务**；
 *  **有返回值的延时任务**；
 *  **固定周期的定时任务**；
 *  **固定延时的定时任务**。

无论是提交哪种类型的任务，`ScheduledThreadPoolExecutor`均会将提交的任务封装为`ScheduledFutureTask`，然后再将`ScheduledFutureTask`添加到延时队列中。

`ScheduledFutureTask`添加到延时队列后，还会同时创建一个初始任务为**null**的`Worker`，并且这个`Worker`会到延时队列中去获取任务。