首页>>后端>>java->java线程池之ScheduledThreadPoolExecutor实现原理

java线程池之ScheduledThreadPoolExecutor实现原理

时间:2023-12-02 本站 点击:0

java中异步周期任务调度有Timer,ScheduledThreadPoolExecutor等实现,目前单机版的定时调度都是使用ScheduledThreadPoolExecutor去实现,那么它是如何实现周期执行任务的呢?其实它还是利用ThreadPoolExecutor线程池去执行任务,这一点从它是继承自ThreadPoolExecutor救可以看的出来,其实关键在于如何实现任务的周期性调度,

ScheduledThreadPoolExecutor类以及核心函数

首先ScheduledThreadPoolExecutor是实现ScheduledExecutorService接口,它主要定义了四个方法:

周期调度一个Runnable的对象

周期调度一个Callable的对象

固定周期调度Runnable对象 (不管上一次Runnable执行结束的时间,总是以固定延迟时间执行 即 上一个Runnable执行开始时候 + 延时时间 = 下一个Runnable执行的时间点)

以固定延迟调度unnable对象(当上一个Runnable执行结束后+固定延迟 = 下一个Runnable执行的时间点)

publicinterfaceScheduledExecutorServiceextendsExecutorService{publicScheduledFuture<?>schedule(Runnablecommand,longdelay,TimeUnitunit);public<V>ScheduledFuture<V>schedule(Callable<V>callable,longdelay,TimeUnitunit);publicScheduledFuture<?>scheduleAtFixedRate(Runnablecommand,longinitialDelay,longperiod,TimeUnitunit);publicScheduledFuture<?>scheduleWithFixedDelay(Runnablecommand,longinitialDelay,longdelay,TimeUnitunit);}

其次,ScheduledThreadPoolExecutor是继承ThreadPoolExecutor,所以它是借助线程池的能力去执行任务,然后自身去实现周期性调度。从构造方法调用父类的线程池的构造方法,核心线程数是构造方法传入,这里可以看到最大线程数是Integer的最大值即2147483647, 还有等待队列是DelayedWorkQueue,它是实现延时的关键.

/***Createsanew{@codeScheduledThreadPoolExecutor}withthe*givencorepoolsize.**@paramcorePoolSizethenumberofthreadstokeepinthepool,even*iftheyareidle,unless{@codeallowCoreThreadTimeOut}isset*@throwsIllegalArgumentExceptionif{@codecorePoolSize<0}*/publicScheduledThreadPoolExecutor(intcorePoolSize){super(corePoolSize,Integer.MAX_VALUE,0,NANOSECONDS,newDelayedWorkQueue());}

scheduleAtFixedRate是实现周期性调度的方法,调度任务就是实现Runnable对象, 以及系统的开始延时时间,周期的调度的间隔时间。

计算初始触发时间和执行周期,并和传入的Runnable对象作为参数封装成 ScheduledFutureTask,然后调用decorateTask装饰Tas(默认实现为空)。

设置ScheduledFutureTask对象outerTask为t(默认就是它自己)。

调用delayedExecute延迟执行任务。

publicScheduledFuture<?>scheduleAtFixedRate(Runnablecommand,longinitialDelay,longperiod,**TimeUnitunit){if(command==null||unit==null)thrownewNullPointerException();if(period<=0)thrownewIllegalArgumentException();ScheduledFutureTask<Void>sft=newScheduledFutureTask<Void>(command,null,triggerTime(initialDelay,unit),unit.toNanos(period));RunnableScheduledFuture<Void>t=decorateTask(command,sft);sft.outerTask=t;delayedExecute(t);returnt;}

判断线程池状态,如果不是处于running状态,则拒绝该任务。

将该任务加入父类的延迟队列(实际为初始化的DelayedWorkQueue对象)

再次判断线程池不是处于running状态,并且,判断是否是处于shutdown状态并且continueExistingPeriodicTasksAfterShutdown标志是否是true(默认是false,表示是否线程次处于shutdown状态下是否继续执行周期性任务),若果为true,则从队列删除任务,false,则确保启动线程来执行周期性任务

privatevoiddelayedExecute(RunnableScheduledFuture<?>task){if(isShutdown())reject(task);else{super.getQueue().add(task);if(isShutdown()&&!canRunInCurrentRunState(task.isPeriodic())&&remove(task))task.cancel(false);elseensurePrestart();}}

获取线程池数量

如果小于核心线程数,则启动核心线程执行任务,如果线程数为空,则启动非核心线程

voidensurePrestart(){intwc=workerCountOf(ctl.get());if(wc<corePoolSize)addWorker(null,true);elseif(wc==0)addWorker(null,false);}

ScheduledFutureTask的run函数

获取是否是周期性任务

判断是否线程池状态是否可以执行任务,如果为true,则取消任务 3 如果是非周期性任务,则直接调用父类FutureTask的run方法, 4 如果是周期性任务,则调用FutureTask的runAndReset函数, 如果该函数返回为true,则调用setNextRunTime设置下一次运行的时间, 并且还行reExecutePeriodic再次执行周期性任务。

publicvoidrun(){booleanperiodic=isPeriodic();if(!canRunInCurrentRunState(periodic))cancel(false);elseif(!periodic)ScheduledFutureTask.super.run();elseif(ScheduledFutureTask.super.runAndReset()){setNextRunTime();reExecutePeriodic(outerTask);}}

判断线程池是否处于可执行任务的状态,如果为true,则重新将设置下一次运行时间的任务加入父类的等待队列,

如果线程池处于不可运行任务的状态,则并且从等待队列中移除成功, 调用任务的取消操作,否则调用ensurePrestart确保启动线程执行任务

voidreExecutePeriodic(RunnableScheduledFuture<?>task){if(canRunInCurrentRunState(true)){super.getQueue().add(task);if(!canRunInCurrentRunState(true)&&remove(task))task.cancel(false);elseensurePrestart();}}

DelayedWorkQueue类核心函数

DelayedWorkQueue是继承AbstractQueue,并实现BlockingQueue接口

staticclassDelayedWorkQueueextendsAbstractQueue<Runnable>implementsBlockingQueue<Runnable>{

核心字段

//初始容量为16privatestaticfinalintINITIAL_CAPACITY=16;//等待队列,只能保存RunnableScheduledFuture对象privateRunnableScheduledFuture<?>[]queue=newRunnableScheduledFuture<?>[INITIAL_CAPACITY];//锁privatefinalReentrantLocklock=newReentrantLock();//对俄大小privateintsize=0;//leader线程,表示最近需要执行的任务的线程。privateThreadleader=null;//条件锁privatefinalConditionavailable=lock.newCondition();

offer函数:

将添加的参数转换成RunnableScheduledFuture对象。

加全局锁。

获取当前队列的size,如果等于队列的长度,则嗲用grow扩容,增加50%的数组长度。

size加1。

如果数组为0,则将加入的对象放在索引为0的位置, 然后设置ScheduledFutureTask的heapIndex的索引(便于后续快速删除)。

调用siftUp做堆的上浮操作,这里是小根堆的操作。

如果队列中第一个元素是传入的对象,则将laader设置null

释放锁

返回true

publicbooleanoffer(Runnablex){if(x==null)thrownewNullPointerException();RunnableScheduledFuture<?>e=(RunnableScheduledFuture<?>)x;finalReentrantLocklock=this.lock;lock.lock();try{inti=size;if(i>=queue.length)grow();size=i+1;if(i==0){queue[0]=e;setIndex(e,0);}else{siftUp(i,e);}if(queue[0]==e){leader=null;available.signal();}}finally{lock.unlock();}returntrue;}

siftUp主要就是做小根堆的上移操作,从if (key.compareTo(e) >= 0) 看出,如果key大于parent索引的元素,则停止。

/***Createsanew{@codeScheduledThreadPoolExecutor}withthe*givencorepoolsize.**@paramcorePoolSizethenumberofthreadstokeepinthepool,even*iftheyareidle,unless{@codeallowCoreThreadTimeOut}isset*@throwsIllegalArgumentExceptionif{@codecorePoolSize<0}*/publicScheduledThreadPoolExecutor(intcorePoolSize){super(corePoolSize,Integer.MAX_VALUE,0,NANOSECONDS,newDelayedWorkQueue());}0

poll函数

加锁

获取队列中索引为0的云元素,若果为null或者第一个元素的执行时间戳时间大于当前时间则直接返回null,否则调用finishPoll将第一个元素返回.

释放锁

/***Createsanew{@codeScheduledThreadPoolExecutor}withthe*givencorepoolsize.**@paramcorePoolSizethenumberofthreadstokeepinthepool,even*iftheyareidle,unless{@codeallowCoreThreadTimeOut}isset*@throwsIllegalArgumentExceptionif{@codecorePoolSize<0}*/publicScheduledThreadPoolExecutor(intcorePoolSize){super(corePoolSize,Integer.MAX_VALUE,0,NANOSECONDS,newDelayedWorkQueue());}1

将队列size 减 1

获取队列中队列中最后一个元素,并且设置队列最后一个为null

最后一个元素不为null,则调用sfitdown进行,将最后一个元素设置到索引为0的位置,将下移操作,重新调整小根堆。

ScheduledFutureTask的heapIndex为-1

/***Createsanew{@codeScheduledThreadPoolExecutor}withthe*givencorepoolsize.**@paramcorePoolSizethenumberofthreadstokeepinthepool,even*iftheyareidle,unless{@codeallowCoreThreadTimeOut}isset*@throwsIllegalArgumentExceptionif{@codecorePoolSize<0}*/publicScheduledThreadPoolExecutor(intcorePoolSize){super(corePoolSize,Integer.MAX_VALUE,0,NANOSECONDS,newDelayedWorkQueue());}2

ScheduledFutureTask的compareTo函数

ScheduledFutureTask实现compareTo方法逻辑

首先比较是否是同一个对象

若果是ScheduledFutureTask对象,则比较time的大小,time是下一次执行的任务的时间戳,如果不是,则比较 getDelay的时间大小

/***Createsanew{@codeScheduledThreadPoolExecutor}withthe*givencorepoolsize.**@paramcorePoolSizethenumberofthreadstokeepinthepool,even*iftheyareidle,unless{@codeallowCoreThreadTimeOut}isset*@throwsIllegalArgumentExceptionif{@codecorePoolSize<0}*/publicScheduledThreadPoolExecutor(intcorePoolSize){super(corePoolSize,Integer.MAX_VALUE,0,NANOSECONDS,newDelayedWorkQueue());}3

ScheduledThreadPoolExecutor的take函数就是ThreadPoolExecutor的从任务队列中获取任务,没有任务则一直等待(这里是线程数小于核心线程数的情况)

加可中断锁

获取队列中第一个元素的任务,从前面可以知道此任务执行的时间戳最小的任务

如果第一个任务为空,则再全局的锁的条件锁上等待,

如果第一个任务不为空,则获取延迟时间,如果延时时间小于0,说明第一个任务已经到时间了,则返回第一个任务。

如果leader线程不为空,则让线程在全局锁的条件锁上等待

如果leader为空,则将获取第一个任务的当前线程赋值为leader变量。

在全局锁的条件锁上等待delay纳秒, 等待结束后,如果当前线程还是等于leader线程,则重置leader为空

最后判断 leader为空并且第一个任务不为空,则唤醒全局锁上条件锁的等待的线程。

释放全局锁。

/***Createsanew{@codeScheduledThreadPoolExecutor}withthe*givencorepoolsize.**@paramcorePoolSizethenumberofthreadstokeepinthepool,even*iftheyareidle,unless{@codeallowCoreThreadTimeOut}isset*@throwsIllegalArgumentExceptionif{@codecorePoolSize<0}*/publicScheduledThreadPoolExecutor(intcorePoolSize){super(corePoolSize,Integer.MAX_VALUE,0,NANOSECONDS,newDelayedWorkQueue());}4

总结\ 综合前面所述,线程池从DelayedWorkQueue每次取出的任务就是延迟时间最小的任务, 若果到达时间的任务,则执行任务,否则则用条件锁Conditon的wait进行等待,执行完后,则用signal进行唤醒下一个任务的执行。


本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。
如若转载,请注明出处:/java/10504.html