Spring源码情操陶冶-任务定时器ConcurrentTaskScheduler

时间:2022-05-12 00:14:32

承接前文Spring源码情操陶冶#task:scheduled-tasks解析器,本文在前文的基础上讲解单核心线程线程池的工作原理

应用附例

承接前文的例子,如下

<!--define bean for schedule task-->
<bean id="taskBean" class="com.jing.test.spring.task.TaskBean"></bean> <task:scheduled-tasks>
<task:scheduled ref="taskBean" method="doInit" cron="0 0 0 ? * *"></task:scheduled>
<task:scheduled ref="taskBean" method="doClear" cron="0 0 23 ? * *"></task:scheduled>
</task:scheduled-tasks>

即我们不配置task:scheduled-tasks的属性scheduler,则会采取org.springframework.scheduling.concurrent.ConcurrentTaskScheduler任务定时器。

实例化缘由

可以直接去看前文的ContextLifecycleScheduledTaskRegistrar#scheduleTasks()的一个代码片段,如下

protected void scheduleTasks() {
****
****
// 如果不指定scheduler属性,则默认使用单线程池模型
if (this.taskScheduler == null) {
this.localExecutor = Executors.newSingleThreadScheduledExecutor();
this.taskScheduler = new ConcurrentTaskScheduler(this.localExecutor);
}
**** }

我们可以观察下该定时器所包含的线程池对象,调用的是JDK Executors的静态方法newSingleThreadScheduledExecutor()方法

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}

直接去看ScheduledThreadPoolExecutor构造函数

    public ScheduledThreadPoolExecutor(int corePoolSize) {
// 应用ThreadPoolExecutor的构造方法,这里很熟悉了
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}

从上面看出,创建的线程池类型为ScheduledThreadPoolExecutor,其内部情况如下

  • 核心线程-1个
  • 最大可创建线程-Integer.MAX_VALUE
  • 空闲线程活动时间-0秒
  • 线程队列-DelayedWorkQueue
  • 拒绝策略-AbortPolicy(抛异常信息)

定时器执行入口

由前文得知,定时器会根据任务的类型执行TaskScheduler接口的scheduleAtFixedRate()或者scheduleWithFixedDelay()抑或是schedule()方法。为了方便理解,我们就拿schedule()方法入手

ConcurrentTaskScheduler#schedule()

	@Override
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
try {
// jdk1.8不支持,jdk1.7支持
if (this.enterpriseConcurrentScheduler) {
return new EnterpriseConcurrentTriggerScheduler().schedule(decorateTask(task, true), trigger);
}
else {
// ErrorHandler一般为LoggingErrorHandler,打印错误日志,也可以直接抛出异常
ErrorHandler errorHandler = (this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));
return new ReschedulingRunnable(task, trigger, this.scheduledExecutor, errorHandler).schedule();
}
}
catch (RejectedExecutionException ex) {
throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
}
}

由上述代码可知,最后包装为ReschedulingRunnable类来调用schedule()方法,OK,在此之前我们必须好好观察下ReschedulingRunnable

ReschedulingRunnable

  1. 首先看下其UML类图

    Spring源码情操陶冶-任务定时器ConcurrentTaskScheduler

  2. 再看下其构造函数

	/**
* @param delegate ScheduledMethodRunnable.class Object
*
* @param trigger 一般为CronTrigger
*
* @param executor ScheduledThreadPoolExecutor.class Object
*
* @param errorHandler 一般为LoggingErrorHandler.class Object
*
*/
public ReschedulingRunnable(Runnable delegate, Trigger trigger, ScheduledExecutorService executor, ErrorHandler errorHandler) {
// 由父类保存真实的Runnable对象
super(delegate, errorHandler);
this.trigger = trigger;
this.executor = executor;
}
  1. 然后直接看下schedule()方法的代码
	public ScheduledFuture<?> schedule() {
synchronized (this.triggerContextMonitor) {
// 计算下一次执行的时间
this.scheduledExecutionTime = this.trigger.nextExecutionTime(this.triggerContext);
if (this.scheduledExecutionTime == null) {
return null;
}
// 由上述的执行时间计算需要延迟的时间,精确到毫秒
long initialDelay = this.scheduledExecutionTime.getTime() - System.currentTimeMillis();
// 由ScheduledThreadPoolExecutor执行定时任务
this.currentFuture = this.executor.schedule(this, initialDelay, TimeUnit.MILLISECONDS);
return this;
}
}
  1. 为了理解方便,我们不直接跳至ScheduledThreadPoolExecutor看源码,我们脑壳想想,肯定会调用其中的run()方法
	@Override
public void run() {
Date actualExecutionTime = new Date();
// 调用委托类DelegatingErrorHandlingRunnable的run方法
super.run();
// 更新时间戳并回调shedule()方法使其能周期性执行任务
Date completionTime = new Date();
synchronized (this.triggerContextMonitor) {
this.triggerContext.update(this.scheduledExecutionTime, actualExecutionTime, completionTime);
if (!this.currentFuture.isCancelled()) {
schedule();
}
}
}

直接看下父类DelegatingErrorHandlingRunnable#run()方法

	@Override
public void run() {
try {
this.delegate.run();
}
catch (UndeclaredThrowableException ex) {
this.errorHandler.handleError(ex.getUndeclaredThrowable());
}
catch (Throwable ex) {
this.errorHandler.handleError(ex);
}
}

上述的代码则会去执行委托类ScheduledMethodRunnablerun()方法,脑壳想想就是通过JDK的method.invoke(Object obj,Object.. args)方法执行我们自定义写的业务。并由errorHandler捕获异常进行善后,默认只是进行error级别的日志输出。

  1. 对此类作下小结
  • 执行定时器最终调用ScheduledThreadPoolExecutor#schedule()方法,并返回ScheduledFuture对象

  • run()方法最终会调用ScheduledMethodRunnable对象来执行用户自定义的业务

  • run()方法也会调用schedule()方法来重复执行第一点的步骤以达到定时执行的效果

ConcurrentTaskScheduler如何保证单线程活跃执行任务

我们从上文得知,其corePoolSize=1,最终原因我们需要从ScheduledThreadPoolExecutor这个继承了ThreadPoolExecutor通用线程池类来了解

ScheduledThreadPoolExecutor#schedule()

直接看下源码

    public ScheduledFuture<?> schedule(Runnable command,
long delay,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
// 直接返回ScheduledFutureTask对象
RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Void>(command, null,
triggerTime(delay, unit)));
// 延迟执行策略,继续追踪
delayedExecute(t);
return t;
}

继续追踪delayedExecute()方法

    private void delayedExecute(RunnableScheduledFuture<?> task) {
// 如果线程池关闭了则直接拒绝任务
if (isShutdown())
reject(task);
else {
// 优先将任务放入DelayQueue队列
super.getQueue().add(task);
// 保护策略,避免添加到队列后线程池关闭了,尝试删除任务
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
else
// 确保已有线程在跑
ensurePrestart();
}
}

继续追踪ensurePrestart()方法

    void ensurePrestart() {
int wc = workerCountOf(ctl.get());
if (wc < corePoolSize)
addWorker(null, true);
else if (wc == 0)
addWorker(null, false);
}

由上述的源码我们可以得知

  1. 当线程池数小于执行的核心线程数corePoolSize,则会创建线程,不管核心与否我们都可以得知一旦当前线程数>=核心线程数,则不会进行新线程的创建

  2. ConcurrentTaskScheduler指定的corePoolSize=1,由第一点得知,永远只有一个线程存在于线程池中

ConcurrentTaskScheduler如何保证任务延迟指定时长后被执行

为了避免大片的源码影响我们的阅读以及理解,博主只在此处指出ConcurrentTaskScheduler所拥有DelayQueue队列的take()方法实现了延迟等待的效果

    public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
// 可重入同步锁,确保该方法线程安全
lock.lockInterruptibly();
try {
for (;;) {
E first = q.peek();
// 队列无任务,那就处于休眠等待
if (first == null)
available.await();
else {
// 获取还需等待的时间
long delay = first.getDelay(NANOSECONDS);
if (delay <= 0)
return q.poll();
first = null; // don't retain ref while waiting
if (leader != null)
available.await();
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// 等待相应的时间后才可放行,采取的是Condition的机制
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// 唤醒等待的线程或者等待的条件
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}

上述代码用到了可重入锁以及锁条件Condition的相关知识,后续会详细分析锁的相关知识。此处作下总结

  1. ReentrantLock锁为可重入锁,即相同的线程可再次获取该锁,不必等待阻塞

  2. ReentrantLockCondition机制,其类似于Object.wait()机制

  3. 其调用available.awaitNanos(delay);方法使当前线程休眠指定的时间后,最终会调用q.poll()方法返回待处理的任务

  4. 有兴趣的读者可自行阅读ThreadPoolExecutorrunWorker()方法和getTask()方法,便可以彻底理解ScheduledThreadPoolExecutor的定时机制

小结

CocurrentTaskScheduler指定的单线程模型,会让任务按照FIFO的机制有序的被执行,这个模型不大适合多个任务的同时定时执行,会导致任务执行有一定的延迟性。所以建议与spring结合时配置task:scheduler并配置pool-size属性

Spring源码情操陶冶-任务定时器ConcurrentTaskScheduler的更多相关文章

  1. Spring源码情操陶冶&num;task&colon;scheduled-tasks解析器

    承接前文Spring源码情操陶冶#task:executor解析器,在前文基础上解析我们常用的spring中的定时任务的节点配置.备注:此文建立在spring的4.2.3.RELEASE版本 附例 S ...

  2. Spring源码情操陶冶-AbstractApplicationContext&num;finishBeanFactoryInitialization

    承接前文Spring源码情操陶冶-AbstractApplicationContext#registerListeners 约定web.xml配置的contextClass为默认值XmlWebAppl ...

  3. Spring源码情操陶冶-AbstractApplicationContext&num;registerListeners

    承接前文Spring源码情操陶冶-AbstractApplicationContext#onRefresh 约定web.xml配置的contextClass为默认值XmlWebApplicationC ...

  4. Spring源码情操陶冶-AbstractApplicationContext&num;onRefresh

    承接前文Spring源码情操陶冶-AbstractApplicationContext#initApplicationEventMulticaster 约定web.xml配置的contextClass ...

  5. Spring源码情操陶冶-AbstractApplicationContext&num;initApplicationEventMulticaster

    承接前文Spring源码情操陶冶-AbstractApplicationContext#initMessageSource 约定web.xml配置的contextClass为默认值XmlWebAppl ...

  6. Spring源码情操陶冶-AbstractApplicationContext&num;initMessageSource

    承接前文Spring源码情操陶冶-AbstractApplicationContext#registerBeanPostProcessors 约定web.xml配置的contextClass为默认值X ...

  7. Spring源码情操陶冶-AbstractApplicationContext&num;registerBeanPostProcessors

    承接前文Spring源码情操陶冶-AbstractApplicationContext#invokeBeanFactoryPostProcessors 瞧瞧官方注释 /** * Instantiate ...

  8. Spring源码情操陶冶-AbstractApplicationContext&num;invokeBeanFactoryPostProcessors

    阅读源码有利于陶冶情操,承接前文Spring源码情操陶冶-AbstractApplicationContext#postProcessBeanFactory 约定:web.xml中配置的context ...

  9. Spring源码情操陶冶-AbstractApplicationContext&num;postProcessBeanFactory

    阅读源码有利于陶冶情操,承接前文Spring源码情操陶冶-AbstractApplicationContext#prepareBeanFactory 约定:web.xml中配置的contextClas ...

随机推荐

  1. js 输出数组最大值

    ,,,,]; ]; ; i < run.length; i++) { if (max<run[i]) { max=run[i]; }else{ max=max; } } alert(max ...

  2. mybatis学习(一) mybatis框架的特性

    mybatis 的源代码地址是https://github.com/mybatis/mybatis-3/ 以及相关文档 All the information i get from http://ww ...

  3. JDBC之事务隔离级别以及ACID特性

    JDBC之事务隔离级别以及ACID特性 事务隔离级别: 1.更新遗失(Lost update) 两个事务都同时更新一行数据,但是第二个事务却中途失败退出,导致对数据的两个修改都失效了.这是因为系统没有 ...

  4. 决策树ID3算法的java实现

    决策树的分类过程和人的决策过程比较相似,就是先挑“权重”最大的那个考虑,然后再往下细分.比如你去看医生,症状是流鼻涕,咳嗽等,那么医生就会根据你的流鼻涕这个权重最大的症状先认为你是感冒,接着再根据你咳 ...

  5. MySql学习&lpar;七&rpar; —— 查询性能优化 深入理解MySql如何执行查询

    本篇深入了解查询优化和服务器的内部机制,了解MySql如何执行特定查询,从中也可以知道如何更改查询执行计划,当我们深入理解MySql如何真正地执行查询,明白高效和低效的真正含义,在实际应用中就能扬长避 ...

  6. 利用python设计PDF报告,jinja2,whtmltopdf,matplotlib,pandas

    转自:https://foofish.net/python-crawler-html2pdf.html 工具准备 弄清楚了网站的基本结构后就可以开始准备爬虫所依赖的工具包了.requests.beau ...

  7. JSP JSTL知识结构图

    自行绘制,欢迎指正.

  8. MySQL 存储过程&sol;游标&sol;事务

    将会用到的几个表 mysql> DESC products; +------------+--------------+------+-----+---------+-------------- ...

  9. tensorflow笔记2:TensorBoard

    Tensorboard中的参数 Summary:所有需要在TensorBoard上展示的统计结果. tf.name_scope():为Graph中的Tensor添加层级,TensorBoard会按照代 ...

  10. 32位MD5加密补齐丢失的0

    /// <summary> /// 获取32位MD5加密字符串(已补完0) /// </summary> /// <param name="strWord&qu ...