热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

深入解析ScheduledThreadPoolExecutor并发执行机制源代码

本文深入剖析了ScheduledThreadPoolExecutor的并发执行机制及其源代码,详细解读了该线程池如何在指定延时或定期执行任务,探讨了其内部的工作原理和优化策略,为开发者提供了宝贵的参考和实践指导。

ScheduledThreadPoolExecutor是一个可以在指定一定延时时间后或者定时进行任务调度的线程池,ScheduledThreadPoolExecutor继承了ThreadPoolExecutor并实现了ScheduledExecutorService接口。线程池的队列是DelayedWorkQueue(他是ScheduledThreadPoolExecutor的一个内部类)。
在这里插入图片描述

还要在看一下ScheduledFutureTask(同样是ScheduledThreadPoolExecutor的一个内部类),继承FutureTask,FutureTask的内部有一个变量state用来表示任务的状态,一开始状态为NEW。下面是所有状态定义。

private volatile int state;private static final int NEW = 0;//初始化状态private static final int COMPLETING = 1;//执行中状态private static final int NORMAL = 2;//正常运行结束状态private static final int EXCEPTIONAL = 3;//运行中异常private static final int CANCELLED = 4;//任务被取消private static final int INTERRUPTING = 5;//任务正在被中段private static final int INTERRUPTED = 6;//任务已经被中断

ScheduledFutureTask内部还有一个变量period用来表示任务的类型,如果==0,则表示任务是一次性的,任务执行完毕后退出,如果为负数,说明当前任务是以固定延时的定时可重复执行任务,如果为正数,说明任务是以固定频率的定时定时可重复执行任务。

一、chedule(Runnable command, long delay, TimeUnit unit)方法解析

他的作用是提交一个延时执行的任务,任务从提交时间算起延时单位为unit的delay时间后开始执行,任务只会执行一次。

public ScheduledFuture<?> schedule(Runnable command,long delay,TimeUnit unit) {//判断参数是否为空&#xff0c;空则跑出异常if (command &#61;&#61; null || unit &#61;&#61; null)throw new NullPointerException();//任务转换RunnableScheduledFuture<?> t &#61; decorateTask(command,new ScheduledFutureTask<Void>(command, null,triggerTime(delay, unit)));//添加任务到延时队列delayedExecute(t);return t;}

首先是参数判断&#xff0c;为空则抛出异常&#xff0c;接着是装饰任务&#xff0c;把提交的command(Runnable对象)转换为ScheduledFutureTask&#xff0c;ScheduledFutureTask是具体放入延时队列里面的东西&#xff0c;由于是延时任务&#xff0c;所以ScheduledFutureTask实现了getDelay和compareTo方法&#xff0c;triggerTime方法将延时时间转换为绝对时间&#xff0c;也就是把当前时间的纳秒加上延迟的纳秒后的值。

ScheduledFutureTask的构造如下&#xff0c;设period为0&#xff0c;表示该任务是一次性任务。

ScheduledFutureTask(Callable<V> callable, long ns) {super(callable);this.time &#61; ns;this.period &#61; 0;this.sequenceNumber &#61; sequencer.getAndIncrement();}

然后通过delayedExecute将任务添加到延时队列。

private void delayedExecute(RunnableScheduledFuture<?> task) {//如果线程池管理&#xff0c;则执行线程拒绝策略if (isShutdown())reject(task);else {//添加任务到延时队列super.getQueue().add(task);//再次检查线程池状态if (isShutdown() &&!canRunInCurrentRunState(task.isPeriodic()) &&remove(task))task.cancel(false);else//确保至少一个线程在处理任务ensurePrestart();}}

上述代码首先确保线程池没关闭&#xff0c;关闭则执行拒绝策略&#xff0c;没关闭将任务添加到延时队列&#xff0c;添加后再重新检查线程池是否关闭&#xff0c;如果关闭则从延时队列里面删除刚才添加的任务。

再看ensurePrestart方法。

void ensurePrestart() {int wc &#61; workerCountOf(ctl.get());//增加核心线程数if (wc < corePoolSize)addWorker(null, true);else if (wc &#61;&#61; 0)addWorker(null, false);}

首先获取到了线程池中的线程数&#xff0c;如果个数小于核心线程池则新增一个线程&#xff0c;否则如果当前线程数为0个&#xff0c;则同样新增一个线程。

我们知道ThreadPoolExecutor在具体执行任务的线程是Worker线程&#xff0c;Worker线程调用具体任务的run方法来执行&#xff0c;在这里的任务是ScheduledFutureTask&#xff0c;所以来看看ScheduledFutureTask的run方法。

public void run() {//是否执行一次boolean periodic &#61; isPeriodic();//取消任务if (!canRunInCurrentRunState(periodic))cancel(false);else if (!periodic)ScheduledFutureTask.super.run();//定时任务else if (ScheduledFutureTask.super.runAndReset()) {setNextRunTime();reExecutePeriodic(outerTask);}
}

首先判断任务是一次性的还是可重复执行的任务&#xff0c;ScheduledFutureTask在构造方法中已经设置了period为0&#xff0c;所以&#xff0c;这里会返回false。

public boolean isPeriodic() {return period !&#61; 0;}

然后判断当前任务是否应该被取消。为true则取消任务。

由于periodic为false&#xff0c;则会执行代码ScheduledFutureTask.super.run();&#xff0c;调用父类FutureTask的run方法。

public void run() {if (state !&#61; NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))return;try {Callable<V> c &#61; callable;if (c !&#61; null && state &#61;&#61; NEW) {V result;boolean ran;try {result &#61; c.call();ran &#61; true;} catch (Throwable ex) {result &#61; null;ran &#61; false;setException(ex);}if (ran)set(result);}} finally {runner &#61; null;int s &#61; state;if (s >&#61; INTERRUPTING)handlePossibleCancellationInterrupt(s);}}

FutureTask.run()首先判断任务状态&#xff0c;如果不是NEW则直接返回&#xff0c;或者如果任务状态为NEW&#xff0c;但是使用CAS设置当前任务的持有者为当前线程失败则直接返回。

然后具体调用callable的call方法执行任务&#xff0c;如果任务执行成功则修改任务状态&#xff0c;也就是set方法。

protected void set(V v) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome &#61; v;UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final statefinishCompletion();}
}

使用CAS将当前任务的状态从NEW转换的COMPLETING。这里当有多个线程调用时只有一个线程会成功&#xff0c;成功的线程在通过 UNSAFE.putOrderedInt设置任务的状态为正常结束状态。

还有在任务执行失败后&#xff0c;执行setException方法&#xff0c;和set方法类似了。

protected void setException(Throwable t) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome &#61; t;UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final statefinishCompletion();}}

二、scheduleWithFixedDelay(Runnable command,long initialDelay, long delay,TimeUnit unit)方法解析

他的作用是&#xff0c;当任务执行完毕后&#xff0c;让其延迟固定时间后再次运行&#xff0c;initialDelay表示提交任务后延迟多少时间开始执行任务command&#xff0c;delay表示当任务执行完毕后延长多少时间后再次运行command&#xff0c;unit是时间单位。

这个任务会一直重复运行下去&#xff0c;直到任务中抛出异常、被取消、线程池关闭。

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit) {//参数判断if (command &#61;&#61; null || unit &#61;&#61; null)throw new NullPointerException();if (delay <&#61; 0)throw new IllegalArgumentException();//任务转换ScheduledFutureTask<Void> sft &#61;new ScheduledFutureTask<Void>(command,null,triggerTime(initialDelay, unit),unit.toNanos(-delay));RunnableScheduledFuture<Void> t &#61; decorateTask(command, sft);sft.outerTask &#61; t;//添加任务到队列delayedExecute(t);return t;}

首先也是参数判断&#xff0c;为空则抛出异常&#xff0c;然后将command任务转换为ScheduledFutureTask&#xff0c;然后添加延迟到队列。

将任务添加到队列后线程池线程会从队列中获取任务&#xff0c;然后调用ScheduledFutureTask的run方法&#xff0c;由于这里period<0&#xff0c;所以isPeriodic返回true&#xff0c;则会执行方法runAndReset()。

protected boolean runAndReset() {if (state !&#61; NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))return false;boolean ran &#61; false;int s &#61; state;try {Callable<V> c &#61; callable;if (c !&#61; null && s &#61;&#61; NEW) {try {c.call(); ran &#61; true;} catch (Throwable ex) {setException(ex);}}} finally {runner &#61; null;s &#61; state;if (s >&#61; INTERRUPTING)handlePossibleCancellationInterrupt(s);}return ran && s &#61;&#61; NEW;
}

他在任务执行完毕后不会设置任务的状态&#xff0c;是为了让任务可重复执行&#xff0c;看最后一句&#xff0c;判断如果当前任务正常执行完毕并且任务状态为NEW则返回true&#xff0c;如果返回true则执行方法setNextRunTime()&#xff0c;用于设置任务下一次的执行时间。

这里p是<0的&#xff0c;然后设置timer为当前时间加上-p&#xff0c;也就是延迟-p时间后再次执行。

private void setNextRunTime() {long p &#61; period;if (p > 0)time &#43;&#61; p;elsetime &#61; triggerTime(-p);}

三、scheduleAtFixedRate(Runnable command, long initialDelay, long period,TimeUnit unit)方法解析

该方法相对起始时间点以固定频率调用指定任务&#xff0c;当把任务提交到线程池并延迟initialDelay时间后开始执行任务command&#xff0c;然后从initialDelay&#43;period时间点再次执行&#xff0c;而后在initialDelay&#43;2*period时间点再次执行&#xff0c;直到抛出异常或者取消、关闭线程池。

原理和scheduleWithFixedDelay类似&#xff0c;我们就看几个不同点&#xff0c;

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit) {if (command &#61;&#61; null || unit &#61;&#61; null)throw new NullPointerException();if (period <&#61; 0)throw new IllegalArgumentException();ScheduledFutureTask<Void> sft &#61;new ScheduledFutureTask<Void>(command,null,triggerTime(initialDelay, unit),unit.toNanos(period));RunnableScheduledFuture<Void> t &#61; decorateTask(command, sft);sft.outerTask &#61; t;delayedExecute(t);return t;}

首先是period&#61;period&#xff0c;不再是-period。所以当前任务执行完毕后调用setNextRunTime设置任务下次执行的时间是 time &#43;&#61; p。

private void setNextRunTime() {long p &#61; period;if (p > 0)time &#43;&#61; p;elsetime &#61; triggerTime(-p);}

如果当前任务还没有执行完&#xff0c;下一次执行任务的时间到了&#xff0c;则不会并发执行&#xff0c;下次要执行的任务会延迟&#xff0c;要等到当前任务执行完毕后再次执行


推荐阅读
author-avatar
航头党员之家
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有