Spring5源码解析之Spring中的异步和计划任务

2023-09-23 03:03:06 155

Java提供了许多创建线程池的方式，并得到一个Future实例来作为任务结果。对于Spring同样小菜一碟，通过其scheduling包就可以做到将任务线程中后台执行。

在本文的第一部分中，我们将讨论下Spring中执行计划任务的一些基础知识。之后，我们将解释这些类是如何一起协作来启动并执行计划任务的。下一部分将介绍计划和异步任务的配置。最后，我们来写个Demo，看看如何通过单元测试来编排计划任务。

什么是Spring中的异步任务？

在我们正式的进入话题之前，对于Spring，我们需要理解下它实现的两个不同的概念：异步任务和调度任务。显然，两者有一个很大的共同点：都在后台工作。但是，它们之间存在了很大差异。调度任务与异步不同，其作用与Linux中的CRONjob完全相同(windows里面也有计划任务)。举个栗子，有一个任务必须每40分钟执行一次，那么，可以通过XML文件或者注解来进行此配置。简单的异步任务在后台执行就好，无需配置执行频率。

因为它们是两种不同的任务类型，它们两个的执行者自然也就不同。第一个看起来有点像Java的并发执行器(concurrencyexecutor)，这里会有专门去写一篇关于Java中的执行器来具体了解。根据Spring文档TaskExecutor所述，它提供了基于Spring的抽象来处理线程池，这点，也可以通过其类的注释去了解。另一个抽象接口是TaskScheduler，它用于在将来给定的时间点来安排任务，并执行一次或定期执行。

在分析源码的过程中，发现另一个比较有趣的点是触发器。它存在两种类型：CronTrigger或PeriodTrigger。第一个模拟CRON任务的行为。所以我们可以在将来确切时间点提交一个任务的执行。另一个触发器可用于定期执行任务。

Spring的异步任务类

让我们从org.springframework.core.task.TaskExecutor类的分析开始。你会发现，其简单的不行，它是一个扩展Java的Executor接口的接口。它的唯一方法也就是是执行，在参数中使用Runnable类型的任务。

packageorg.springframework.core.task; importjava.util.concurrent.Executor; /** *Simpletaskexecutorinterfacethatabstractstheexecution *ofa{@linkRunnable}. * *

Implementationscanuseallsortsofdifferentexecutionstrategies, *suchas:synchronous,asynchronous,usingathreadpool,andmore. * *

EquivalenttoJDK1.5's{@linkjava.util.concurrent.Executor} *interface;extendingitnowinSpring3.0,sothatclientsmaydeclare *adependencyonanExecutorandreceiveanyTaskExecutorimplementation. *ThisinterfaceremainsseparatefromthestandardExecutorinterface *mainlyforbackwardscompatibilitywithJDK1.4inSpring2.x. * *@authorJuergenHoeller *@since2.0 *@seejava.util.concurrent.Executor */ @FunctionalInterface publicinterfaceTaskExecutorextendsExecutor{ /** *Executethegiven{@codetask}. *

Thecallmightreturnimmediatelyiftheimplementationuses *anasynchronousexecutionstrategy,ormightblockinthecase *ofsynchronousexecution. *@paramtaskthe{@codeRunnable}toexecute(never{@codenull}) *@throwsTaskRejectedExceptionifthegiventaskwasnotaccepted */ @Override voidexecute(Runnabletask); }

相对来说，org.springframework.scheduling.TaskScheduler接口就有点复杂了。它定义了一组以schedule开头的名称的方法允许我们定义将来要执行的任务。所有schedule*方法返回java.util.concurrent.ScheduledFuture实例。Spring5中对scheduleAtFixedRate方法做了进一步的充实，其实最终调用的还是ScheduledFuturescheduleAtFixedRate(Runnabletask,longperiod);

publicinterfaceTaskScheduler{ /** *Schedulethegiven{@linkRunnable},invokingitwheneverthetrigger *indicatesanextexecutiontime. *

Executionwillendoncetheschedulershutsdownorthereturned *{@linkScheduledFuture}getscancelled. *@paramtasktheRunnabletoexecutewheneverthetriggerfires *@paramtriggeranimplementationofthe{@linkTrigger}interface, *e.g.a{@linkorg.springframework.scheduling.support.CronTrigger}object *wrappingacronexpression *@returna{@linkScheduledFuture}representingpendingcompletionofthetask, *or{@codenull}ifthegivenTriggerobjectneverfires(i.e.returns *{@codenull}from{@linkTrigger#nextExecutionTime}) *@throwsorg.springframework.core.task.TaskRejectedExceptionifthegiventaskwasnotaccepted *forinternalreasons(e.g.apooloverloadhandlingpolicyorapoolshutdowninprogress) *@seeorg.springframework.scheduling.support.CronTrigger */ @Nullable ScheduledFutureschedule(Runnabletask,Triggertrigger); /** *Schedulethegiven{@linkRunnable},invokingitatthespecifiedexecutiontime. *

Executionwillendoncetheschedulershutsdownorthereturned *{@linkScheduledFuture}getscancelled. *@paramtasktheRunnabletoexecutewheneverthetriggerfires *@paramstartTimethedesiredexecutiontimeforthetask *(ifthisisinthepast,thetaskwillbeexecutedimmediately,i.e.assoonaspossible) *@returna{@linkScheduledFuture}representingpendingcompletionofthetask *@throwsorg.springframework.core.task.TaskRejectedExceptionifthegiventaskwasnotaccepted *forinternalreasons(e.g.apooloverloadhandlingpolicyorapoolshutdowninprogress) *使用了默认实现，值得我们学习使用的，Java9中同样可以有私有实现的，从这里我们可以做到我只通过*一个接口你来实现，我把其他相应的功能默认实现下，最后调用你自定义实现的接口即可，使接口功能更*加一目了然 *@since5.0 *@see#schedule(Runnable,Date) */ defaultScheduledFutureschedule(Runnabletask,InstantstartTime){ returnschedule(task,Date.from(startTime)); } /** *Schedulethegiven{@linkRunnable},invokingitatthespecifiedexecutiontime. *

Executionwillendoncetheschedulershutsdownorthereturned *{@linkScheduledFuture}getscancelled. *@paramtasktheRunnabletoexecutewheneverthetriggerfires *@paramstartTimethedesiredfirstexecutiontimeforthetask *(ifthisisinthepast,thetaskwillbeexecutedimmediately,i.e.assoonaspossible) *@paramperiodtheintervalbetweensuccessiveexecutionsofthetask *@returna{@linkScheduledFuture}representingpendingcompletionofthetask *@throwsorg.springframework.core.task.TaskRejectedExceptionifthegiventaskwasnotaccepted *forinternalreasons(e.g.apooloverloadhandlingpolicyorapoolshutdowninprogress) *@since5.0 *@see#scheduleAtFixedRate(Runnable,Date,long) */ defaultScheduledFuturescheduleAtFixedRate(Runnabletask,InstantstartTime,Durationperiod){ returnscheduleAtFixedRate(task,Date.from(startTime),period.toMillis()); } /** *Schedulethegiven{@linkRunnable},invokingitatthespecifiedexecutiontime *andsubsequentlywiththegivenperiod. *

之前提到两个触发器组件，都实现了org.springframework.scheduling.Trigger接口。这里，我们只需关注一个的方法nextExecutionTime，其定义下一个触发任务的执行时间。它的两个实现，CronTrigger和PeriodicTrigger，由org.springframework.scheduling.TriggerContext来实现信息的存储，由此，我们可以很轻松获得一个任务的最后一个执行时间(lastScheduledExecutionTime)，给定任务的最后完成时间(lastCompletionTime)或最后一个实际执行时间(lastActualExecutionTime)。接下来，我们通过阅读源代码来简单的了解下这些东西。org.springframework.scheduling.concurrent.ConcurrentTaskScheduler包含一个私有类EnterpriseConcurrentTriggerScheduler。在这个class里面，我们可以找到schedule方法：

publicScheduledFutureschedule(Runnabletask,finalTriggertrigger){
ManagedScheduledExecutorServiceexecutor=(ManagedScheduledExecutorService)scheduledExecutor;
returnexecutor.schedule(task,newjavax.enterprise.concurrent.Trigger(){
@Override
publicDategetNextRunTime(LastExecutionle,DatetaskScheduledTime){
returntrigger.nextExecutionTime(le!=null?
newSimpleTriggerContext(le.getScheduledStart(),le.getRunStart(),le.getRunEnd()):
newSimpleTriggerContext());
}
@Override
publicbooleanskipRun(LastExecutionlastExecution,DatescheduledRunTime){
returnfalse;
}
});
}

SimpleTriggerContext从其名字就可以看到很多东西了，因为它实现了TriggerContext接口。

/**
*Simpledataholderimplementationofthe{@linkTriggerContext}interface.
*
*@authorJuergenHoeller
*@since3.0
*/
publicclassSimpleTriggerContextimplementsTriggerContext{
@Nullable
privatevolatileDatelastScheduledExecutionTime;
@Nullable
privatevolatileDatelastActualExecutionTime;
@Nullable
privatevolatileDatelastCompletionTime;
...
/**
*CreateaSimpleTriggerContextwiththegiventimevalues.
*@paramlastScheduledExecutionTimelastscheduledexecutiontime
*@paramlastActualExecutionTimelastactualexecutiontime
*@paramlastCompletionTimelastcompletiontime
*/
publicSimpleTriggerContext(DatelastScheduledExecutionTime,DatelastActualExecutionTime,DatelastCompletionTime){
this.lastScheduledExecutionTime=lastScheduledExecutionTime;
this.lastActualExecutionTime=lastActualExecutionTime;
this.lastCompletionTime=lastCompletionTime;
}
...
}

也正如你看到的，在构造函数中设置的时间值来自javax.enterprise.concurrent.LastExecution的实现，其中：

getScheduledStart：返回上次开始执行任务的时间。它对应于TriggerContext的lastScheduledExecutionTime属性。
getRunStart：返回给定任务开始运行的时间。在TriggerContext中，它对应于lastActualExecutionTime。
getRunEnd：任务终止时返回。它用于在TriggerContext中设置lastCompletionTime。

Spring调度和异步执行中的另一个重要类是org.springframework.core.task.support.TaskExecutorAdapter。它是一个将java.util.concurrent.Executor作为Spring基本的执行器的适配器(描述的有点拗口,看下面代码就明了了)，之前已经描述了TaskExecutor。实际上，它引用了Java的ExecutorService，它也是继承了Executor接口。此引用用于完成所有提交的任务。

/**
*AdapterthattakesaJDK{@codejava.util.concurrent.Executor}and
*exposesaSpring{@linkorg.springframework.core.task.TaskExecutor}forit.
*Alsodetectsanextended{@codejava.util.concurrent.ExecutorService从此解释上面的说明},adapting
*the{@linkorg.springframework.core.task.AsyncTaskExecutor}interfaceaccordingly.
*
*@authorJuergenHoeller
*@since3.0
*@seejava.util.concurrent.Executor
*@seejava.util.concurrent.ExecutorService
*@seejava.util.concurrent.Executors
*/
publicclassTaskExecutorAdapterimplementsAsyncListenableTaskExecutor{
privatefinalExecutorconcurrentExecutor;
@Nullable
privateTaskDecoratortaskDecorator;
...
/**
*CreateanewTaskExecutorAdapter,
*usingthegivenJDKconcurrentexecutor.
*@paramconcurrentExecutortheJDKconcurrentexecutortodelegateto
*/
publicTaskExecutorAdapter(ExecutorconcurrentExecutor){
Assert.notNull(concurrentExecutor,"Executormustnotbenull");
this.concurrentExecutor=concurrentExecutor;
}
/**
*DelegatestothespecifiedJDKconcurrentexecutor.
*@seejava.util.concurrent.Executor#execute(Runnable)
*/
@Override
publicvoidexecute(Runnabletask){
try{
doExecute(this.concurrentExecutor,this.taskDecorator,task);
}
catch(RejectedExecutionExceptionex){
thrownewTaskRejectedException(
"Executor["+this.concurrentExecutor+"]didnotaccepttask:"+task,ex);
}
}
@Override
publicvoidexecute(Runnabletask,longstartTimeout){
execute(task);
}
@Override
publicFuturesubmit(Runnabletask){
try{
if(this.taskDecorator==null&&this.concurrentExecutorinstanceofExecutorService){
return((ExecutorService)this.concurrentExecutor).submit(task);
}
else{
FutureTask

Spring5源码解析之Spring中的异步和计划任务

热门推荐

随机推荐