1、使用Spring定时器任务实现
package com.suyun.modules.vehicle.timetask;
import com.alibaba.schedulerx.worker.domain.JobContext;
import com.alibaba.schedulerx.worker.processor.JavaProcessor;
import com.alibaba.schedulerx.worker.processor.ProcessResult;
import com.suyun.modules.vehicle.timetask.service.CmdDownStatusTaskService;
import com.suyun.vehicle.dao.ManualTaskDto;
import com.suyun.vehicle.service.ManualTaskService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* Description：队列状态，任务状态，进度刷新
* 包括UDS和终端任务队列
*
* @Author： leo.xiong
* @CreateDate： 2020/12/26 17:39
* @Email： leo.xiong@suyun360.com
* @Since：
*/
@Component
public class CmdDownStatusTask extends JavaProcessor implements ManualTaskService.SubManualTaskService {
private static final Logger LOGGER = LoggerFactory.getLogger(CmdDownStatusTask.class);
@Autowired
private CmdDownStatusTaskService cmdDownStatusTaskService;
@Resource(name = "redisTemplate")
private ValueOperations<String, String> valueOperations;
@Value("${schedulerx.test.endpoint:}")
private String testEndpoint;
@Value("${schedulerx.group.id}")
private String groupId;
/**
* 阿里云调度器
*
* @param context
* @return
*/
@Override
public ProcessResult process(JobContext context) {
return processing(context, null);
}
/**
* Spring注解调度器
*
* @return
*/
@Override
@Scheduled(cron = "0 */1 * * * ?")
public void process() {
processing(JobContext.newBuilder().setGroupId(groupId).build(), testEndpoint);
}
/**
* 自动下发状态刷新（每一分钟执行一次）
*/
@Override
public ProcessResult processing(Object context, String testEndpoint) {
ManualTaskDto manualTaskDto = ManualTaskDto.builder().logger(LOGGER).valueOperations(valueOperations).groupId(groupId).testEndpoint(testEndpoint);
Object obj = execute(manualTaskDto, () -> {
return cmdDownStatusTaskService.doCmdDownStatusTask();
});
return obj instanceof Boolean ? new ProcessResult((Boolean) obj) : (obj instanceof ProcessResult ? (ProcessResult) obj : new ProcessResult(true));
}
}
2、Spring定时调度器生效原理
2.1、初始化定时器配置
Spring定义的钩子ScheduledAnnotationBeanPostProcessor
实现了BeanPostProcessor，SmartInitializingSingleton，ApplicationListener接口；
BeanPostProcessor：会对所有Bean，一般是用@Component注解的Bean对象扫描
@Override
public Object postProcessAfterInitialization(final Object bean, String beanName) {
Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
//第一次扫描后，为空的类下次不会再扫描
if (!this.nonAnnotatedClasses.contains(targetClass)) {
//找到当前类添加了Scheduled注解的所有方法
Map<Method, Set<Scheduled>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
(MethodIntrospector.MetadataLookup<Set<Scheduled>>) method -> {
Set<Scheduled> scheduledMethods = AnnotatedElementUtils.getMergedRepeatableAnnotations(
method, Scheduled.class, Schedules.class);
return (!scheduledMethods.isEmpty() ? scheduledMethods : null);
});
//
if (annotatedMethods.isEmpty()) {
this.nonAnnotatedClasses.add(targetClass);
if (logger.isTraceEnabled()) {
logger.trace("No @Scheduled annotations found on bean class: " + bean.getClass());
}
}
else {
// Non-empty set of methods
annotatedMethods.forEach((method, scheduledMethods) ->
scheduledMethods.forEach(scheduled ->
//校验一些方法配置是否正确，如，定时任务只能是无参的方法等
processScheduled(scheduled, method, bean)));
if (logger.isDebugEnabled()) {
logger.debug(annotatedMethods.size() + " @Scheduled methods processed on bean '" + beanName +
"': " + annotatedMethods);
}
}
}
return bean;
}
protected void processScheduled(Scheduled scheduled, Method method, Object bean) {
try {
//方法无参
Assert.isTrue(method.getParameterCount() == 0,
"Only no-arg methods may be annotated with @Scheduled");
//获取注解所在的bean和方法
Method invocableMethod = AopUtils.selectInvocableMethod(method, bean.getClass());
//定义一个Runnable对象，传入对象和方法，这里没有传入参数，这也是注解不能有入参的原因，后面会用反射的invoke执行定时方法
Runnable runnable = new ScheduledMethodRunnable(bean, invocableMethod);
boolean processedSchedule = false;
//
String errorMessage =
"Exactly one of the 'cron', 'fixedDelay(String)', or 'fixedRate(String)' attributes is required";
Set<ScheduledTask> tasks = new LinkedHashSet<>(4);
// Determine initial delay
long initialDelay = scheduled.initialDelay();
String initialDelayString = scheduled.initialDelayString();
if (StringUtils.hasText(initialDelayString)) {
//initialDelay和initialDelayString不能同时出现
Assert.isTrue(initialDelay < 0, "Specify 'initialDelay' or 'initialDelayString', not both");
if (this.embeddedValueResolver != null) {
initialDelayString = this.embeddedValueResolver.resolveStringValue(initialDelayString);
}
if (StringUtils.hasLength(initialDelayString)) {
try {
initialDelay = parseDelayAsLong(initialDelayString);
}
catch (RuntimeException ex) {
//initialDelayString不是有效的Long或者Duration.parse
throw new IllegalArgumentException(
"Invalid initialDelayString value \"" + initialDelayString + "\" - cannot parse into long");
}
}
}
// Check cron expression
String cron = scheduled.cron();
if (StringUtils.hasText(cron)) {
String zone = scheduled.zone();
if (this.embeddedValueResolver != null) {
cron = this.embeddedValueResolver.resolveStringValue(cron);
zone = this.embeddedValueResolver.resolveStringValue(zone);
}
if (StringUtils.hasLength(cron)) {
// cron不支持initialDelay
Assert.isTrue(initialDelay == -1, "'initialDelay' not supported for cron triggers");
processedSchedule = true;
TimeZone timeZone;
if (StringUtils.hasText(zone)) {
timeZone = StringUtils.parseTimeZoneString(zone);
}
else {
timeZone = TimeZone.getDefault();
}
//加入队列，把Runable和定时器规则包装成CronTask对象
tasks.add(this.registrar.scheduleCronTask(new CronTask(runnable, new CronTrigger(cron, timeZone))));
}
}
// At this point we don't need to differentiate between initial delay set or not anymore
if (initialDelay < 0) {
initialDelay = 0;
}
// Check fixed delay
long fixedDelay = scheduled.fixedDelay();
if (fixedDelay >= 0) {
Assert.isTrue(!processedSchedule, errorMessage);
processedSchedule = true;
tasks.add(this.registrar.scheduleFixedDelayTask(new FixedDelayTask(runnable, fixedDelay, initialDelay)));
}
String fixedDelayString = scheduled.fixedDelayString();
if (StringUtils.hasText(fixedDelayString)) {
if (this.embeddedValueResolver != null) {
fixedDelayString = this.embeddedValueResolver.resolveStringValue(fixedDelayString);
}
if (StringUtils.hasLength(fixedDelayString)) {
Assert.isTrue(!processedSchedule, errorMessage);
processedSchedule = true;
try {
fixedDelay = parseDelayAsLong(fixedDelayString);
}
catch (RuntimeException ex) {
//fixedDelayString不是有效的Long或者Duration.parse
throw new IllegalArgumentException(
"Invalid fixedDelayString value \"" + fixedDelayString + "\" - cannot parse into long");
}
tasks.add(this.registrar.scheduleFixedDelayTask(new FixedDelayTask(runnable, fixedDelay, initialDelay)));
}
}
// Check fixed rate
long fixedRate = scheduled.fixedRate();
if (fixedRate >= 0) {
Assert.isTrue(!processedSchedule, errorMessage);
processedSchedule = true;
tasks.add(this.registrar.scheduleFixedRateTask(new FixedRateTask(runnable, fixedRate, initialDelay)));
}
String fixedRateString = scheduled.fixedRateString();
if (StringUtils.hasText(fixedRateString)) {
if (this.embeddedValueResolver != null) {
fixedRateString = this.embeddedValueResolver.resolveStringValue(fixedRateString);
}
if (StringUtils.hasLength(fixedRateString)) {
Assert.isTrue(!processedSchedule, errorMessage);
processedSchedule = true;
try {
fixedRate = parseDelayAsLong(fixedRateString);
}
catch (RuntimeException ex) {
//fixedRateString不是有效的Long或者Duration.parse
throw new IllegalArgumentException(
"Invalid fixedRateString value \"" + fixedRateString + "\" - cannot parse into long");
}
tasks.add(this.registrar.scheduleFixedRateTask(new FixedRateTask(runnable, fixedRate, initialDelay)));
}
}
// Check whether we had any attribute set
Assert.isTrue(processedSchedule, errorMessage);
// Finally register the scheduled tasks
synchronized (this.scheduledTasks) {
//bean任务放入scheduledTasks任务队列中，每个bean维护一个队列，一个类可以有多个@Scheduled方法
Set<ScheduledTask> registeredTasks = this.scheduledTasks.get(bean);
if (registeredTasks == null) {
registeredTasks = new LinkedHashSet<>(4);
this.scheduledTasks.put(bean, registeredTasks);
}
//注册所有的任务
registeredTasks.addAll(tasks);
}
}
catch (IllegalArgumentException ex) {
//方法错误
throw new IllegalStateException(
"Encountered invalid @Scheduled method '" + method.getName() + "': " + ex.getMessage());
}
}
SmartInitializingSingleton：会调用afterSingletonsInstantiated方法，在
ApplicationContext 还未初始化完成，就先注入定时任务，主要是为了尽可能早的注入任务信息
ApplicationListener：监听程序启动成功之后，调用onApplicationEvent方法，执行顺序在SmartInitializingSingleton后面
2.2、执行定时任务
private void finishRegistration() {
if (this.scheduler != null) {
this.registrar.setScheduler(this.scheduler);
}
if (this.beanFactory instanceof ListableBeanFactory) {
Map<String, SchedulingConfigurer> beans =
((ListableBeanFactory) this.beanFactory).getBeansOfType(SchedulingConfigurer.class);
List<SchedulingConfigurer> configurers = new ArrayList<>(beans.values());
AnnotationAwareOrderComparator.sort(configurers);
for (SchedulingConfigurer configurer : configurers) {
//默认单线程线程池，重写可以使用多线程线程池
configurer.configureTasks(this.registrar);
}
}
//存在任务，scheduler线程池为空
if (this.registrar.hasTasks() && this.registrar.getScheduler() == null) {
Assert.state(this.beanFactory != null, "BeanFactory must be set to find scheduler by type");
try {
//根据类型TaskScheduler，从容器里面获取bean，为空抛出异常，之后抓取异常
this.registrar.setTaskScheduler(resolveSchedulerBean(beanFactory, TaskScheduler.class, false));
}
catch (NoUniqueBeanDefinitionException ex) {
//如果根据类型存在多个bean，在根据类型名称找一次，注入名称为taskScheduler，类型为TaskScheduler的bean
logger.debug("Could not find unique TaskScheduler bean", ex);
try {
this.registrar.setTaskScheduler(resolveSchedulerBean(beanFactory, TaskScheduler.class, true));
}
catch (NoSuchBeanDefinitionException ex2) {
if (logger.isInfoEnabled()) {
logger.info("More than one TaskScheduler bean exists within the context, and " +
"none is named 'taskScheduler'. Mark one of them as primary or name it 'taskScheduler' " +
"(possibly as an alias); or implement the SchedulingConfigurer interface and call " +
"ScheduledTaskRegistrar#setScheduler explicitly within the configureTasks() callback: " +
ex.getBeanNamesFound());
}
}
}
catch (NoSuchBeanDefinitionException ex) {
//如果找不到类型TaskScheduler的bean，在根据类型ScheduledExecutorService在容器中查找，否则抛出异常
logger.debug("Could not find default TaskScheduler bean", ex);
// Search for ScheduledExecutorService bean next...
try {
this.registrar.setScheduler(resolveSchedulerBean(beanFactory, ScheduledExecutorService.class, false));
}
catch (NoUniqueBeanDefinitionException ex2) {
//如果存在多个ScheduledExecutorService类型的bean，在根据ScheduledExecutorService和名称taskScheduler找一下
logger.debug("Could not find unique ScheduledExecutorService bean", ex2);
try {
this.registrar.setScheduler(resolveSchedulerBean(beanFactory, ScheduledExecutorService.class, true));
}
catch (NoSuchBeanDefinitionException ex3) {
if (logger.isInfoEnabled()) {
logger.info("More than one ScheduledExecutorService bean exists within the context, and " +
"none is named 'taskScheduler'. Mark one of them as primary or name it 'taskScheduler' " +
"(possibly as an alias); or implement the SchedulingConfigurer interface and call " +
"ScheduledTaskRegistrar#setScheduler explicitly within the configureTasks() callback: " +
ex2.getBeanNamesFound());
}
}
}
catch (NoSuchBeanDefinitionException ex2) {
// 打印异常
logger.debug("Could not find default ScheduledExecutorService bean", ex2);
// Giving up -> falling back to default scheduler within the registrar...
logger.info("No TaskScheduler/ScheduledExecutorService bean found for scheduled processing");
}
}
}
this.registrar.afterPropertiesSet();
}
2.3、4种任务，这里找下Cron任务**
protected void scheduleTasks() {
//没有定义线程池
if (this.taskScheduler == null) {
this.localExecutor = Executors.newSingleThreadScheduledExecutor();
//初始化一个单线程定时调度线程池，包装成ConcurrentTaskScheduler，自定义线程池，也会包装成ConcurrentTaskScheduler对象
this.taskScheduler = new ConcurrentTaskScheduler(this.localExecutor);
}
if (this.triggerTasks != null) {
for (TriggerTask task : this.triggerTasks) {
addScheduledTask(scheduleTriggerTask(task));
}
}
if (this.cronTasks != null) {
for (CronTask task : this.cronTasks) {
addScheduledTask(scheduleCronTask(task));
}
}
if (this.fixedRateTasks != null) {
for (IntervalTask task : this.fixedRateTasks) {
addScheduledTask(scheduleFixedRateTask(task));
}
}
if (this.fixedDelayTasks != null) {
for (IntervalTask task : this.fixedDelayTasks) {
addScheduledTask(scheduleFixedDelayTask(task));
}
}
}
@Nullable
public ScheduledTask scheduleCronTask(CronTask task) {
//根据任务对象，移除任务，并返回前面的任务，如果没有就新建一个任务
ScheduledTask scheduledTask = this.unresolvedTasks.remove(task);
boolean newTask = false;
if (scheduledTask == null) {
scheduledTask = new ScheduledTask(task);
newTask = true;
}
//存在线程池，用线程池定时执行，即使没有taskScheduler，也会初始化一个单线程池，所以taskScheduler 不为空
if (this.taskScheduler != null) {
//把结果绑定对象
scheduledTask.future = this.taskScheduler.schedule(task.getRunnable(), task.getTrigger());
}
else {
addCronTask(task);
this.unresolvedTasks.put(task, scheduledTask);
}
return (newTask ? scheduledTask : null);
}
2.4、执行ConcurrentTaskScheduler的schedule方法
@Override
@Nullable
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
try {
if (this.enterpriseConcurrentScheduler) {
return new EnterpriseConcurrentTriggerScheduler().schedule(decorateTask(task, true), trigger);
}
else {
ErrorHandler errorHandler =
(this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));
//ReschedulingRunnable对象，实现了Runnable和Future接口表明用线程执行，会有线程返回值
return new ReschedulingRunnable(task, trigger, this.scheduledExecutor, errorHandler).schedule();
}
}
catch (RejectedExecutionException ex) {
throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);
}
}
@Nullable
public ScheduledFuture<?> schedule() {
//同一个ReschedulingRunnable执行，线程安全
synchronized (this.triggerContextMonitor) {
//根据最后的时间，获取下次执行时间
this.scheduledExecutionTime = this.trigger.nextExecutionTime(this.triggerContext);
if (this.scheduledExecutionTime == null) {
return null;
}
//下次执行时间-当前系统时间等于需要延迟的时间，如果下次执行时间比当前时间小，可能使用的是单线程，或者，线程任务执行时间过长，结果可能为负数
long initialDelay = this.scheduledExecutionTime.getTime() - System.currentTimeMillis();
//执行ScheduledThreadPoolExecutor的schedule方法
this.currentFuture = this.executor.schedule(this, initialDelay, TimeUnit.MILLISECONDS);
return this;
}
}
public ScheduledFuture<?> schedule(Runnable command,
long delay,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
RunnableScheduledFuture<?> t = decorateTask(command,
//包装一个ScheduledFutureTask对象，实现了Runnable和Future，command也实现了Runnable和Future，被包装为Callable，后面会调用
new ScheduledFutureTask<Void>(command, null,
//触发时间，如果小于0偏移时间就为0，之后加上偏移时间
triggerTime(delay, unit)));
//延迟执行任务
delayedExecute(t);
return t;
}
private void delayedExecute(RunnableScheduledFuture<?> task) {
//如果线程池状态为shutdown，则拒绝新增任务
if (isShutdown())
reject(task);
else {
//任务加入队列
super.getQueue().add(task);
//如果线程池状态shutdown，并且运行的线程不是循环执行（定时执行），并且可以移除队列任务，那么移除之后终止任务，否则开始执行
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
else
//执行任务
ensurePrestart();
}
}
void ensurePrestart() {
int wc = workerCountOf(ctl.get());
//线程池中正在执行的任务个数
if (wc < corePoolSize)
//如果执行任务个数小于核心线程，进入线程池执行方法
addWorker(null, true);
else if (wc == 0)
//如果线程执行任务为空，进入执行方法
addWorker(null, false);
}
2.5、调用线程池执行传入的Runable接口，并回收资源
从真正开始执行任务（执行线程，都需要执行当前方法）代码一样
3、ScheduledThreadPoolExecutor.ScheduledFutureTask开始启动
- 实现了Comparable 接口，比较的对象为Delayed
- ScheduledThreadPoolExecutor线程池维护的是一个DelayedWorkQueue优先级自增长队列，最大容量为Integer.MAX_VALUE，是一个堆，堆首放的就是时间最早执行的任务，根据Delayed比较时间，每次新增，删除，执行后都会更新下一次任务执行时间，更新堆信息；
public void run() {
//判断是否是周期任务
boolean periodic = isPeriodic();
//线程池非运行状态（RUNNING || (SHUTDOWN && 可以终止)）
if (!canRunInCurrentRunState(periodic))
//取消任务，回收资源
cancel(false);
else if (!periodic)
//非周期性任务，直接执行run
ScheduledFutureTask.super.run();
else if (ScheduledFutureTask.super.runAndReset()) {
//周期性任务执行成功，设置下一次任务时间
setNextRunTime();
//重新入队，调用执行
reExecutePeriodic(outerTask);
}
}
protected boolean runAndReset() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return false;
boolean ran = false;
int s = state;
try {
//callable就是前面定义的ReschedulingRunnable包装的Callable对象
Callable<V> c = callable;
if (c != null && s == NEW) {
try {
c.call(); // don't set result
//如果执行失败，跳入catch，异常被抓取，不会抛出异常
ran = true;
} catch (Throwable ex) {
setException(ex);
}
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
//抓取异常之后，返回false，不会设置下次任务时间，所以在代码层面需要手动抓取异常
return ran && s == NEW;
}
ReschedulingRunnable执行，调用的是
@Override
public void run() {
Date actualExecutionTime = new Date();
//调用的是父级的run方法，实际调用的是this.delegate.run();，delegate就是CornTask获取的Runable，在@Scheuler注解的时候生成的ScheduledMethodRunnable对象
super.run();
Date completionTime = new Date();
synchronized (this.triggerContextMonitor) {
Assert.state(this.scheduledExecutionTime != null, "No scheduled execution");
this.triggerContext.update(this.scheduledExecutionTime, actualExecutionTime, completionTime);
if (!obtainCurrentFuture().isCancelled()) {
schedule();
}
}
}
ScheduledMethodRunnable的run方法
@Override
public void run() {
try {
//设置定时任务方法为可执行
ReflectionUtils.makeAccessible(this.method);
//反射执行定义的定时任务方法
this.method.invoke(this.target);
}
catch (InvocationTargetException ex) {
ReflectionUtils.rethrowRuntimeException(ex.getTargetException());
}
catch (IllegalAccessException ex) {
throw new UndeclaredThrowableException(ex);
}
}
定时任务重新调用
void reExecutePeriodic(RunnableScheduledFuture<?> task) {
if (canRunInCurrentRunState(true)) {
//线程池运行状态，任务重新入队
super.getQueue().add(task);
//线程池不是运行状态，移除任务，结束任务，回收资源
if (!canRunInCurrentRunState(true) && remove(task))
task.cancel(false);
else
//再次调用前置执行方法，递归了
ensurePrestart();
}
}
