SpringBoot Schedule调度任务的动态管理方法是什么(schedule,springboot,开发技术)

前言

定时任务动态管理分为两种方式：

方式一:Web前台配置Trigger触发器(关联Cron)、ThreadPoolTaskScheduler类创建Scheduler方式下进行Schedule调度任务的动态管理

方式二:基于已创建的Schedule调度任务的动态管理，即以组件类 @Scheduled注解声明Schedule调度，在启动程序前一次性初始化，如：

@ComponentpublicclassTestTask{privateDateTimeFormatterdf=DateTimeFormatter.ofPattern("yyyy-MM-ddHH:mm:ss");@Scheduled(cron="0/2****?")publicvoidrobReceiveExpireTask(){System.out.println(df.format(LocalDateTime.now())+"测试测试");}}

缺陷：目前无法在运行期间增加Schedule以及stop、Start、Reset等管理。

一、架构流程图

二、代码实现流程

架构为SpringBoot + Spring + mybatis-plus

1.引入库

pom.xml

<?xmlversion="1.0"encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>merak-hyper-automation-boot</artifactId><groupId>com.merak.automation</groupId><version>1.0.0</version></parent><modelVersion>4.0.0</modelVersion><artifactId>automation-quartz</artifactId><packaging>jar</packaging><repositories><repository><id>aliyun</id><name>aliyunRepository</name><url>http://maven.aliyun.com/nexus/content/groups/public</url><snapshots><enabled>false</enabled></snapshots></repository></repositories><dependencies><!--Spring框架基本的核心工具--><dependency><groupId>org.springframework</groupId><artifactId>spring-context-support</artifactId></dependency><!--SpringWeb模块--><dependency><groupId>org.springframework</groupId><artifactId>spring-web</artifactId></dependency><!--mysql--><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency><!--druid数据连接池--><dependency><groupId>com.alibaba</groupId><artifactId>druid-spring-boot-starter</artifactId></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId></dependency><dependency><groupId>jakarta.validation</groupId><artifactId>jakarta.validation-api</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId></dependency><dependency><groupId>commons-io</groupId><artifactId>commons-io</artifactId></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-annotations</artifactId></dependency><!--引入quartz定时框架--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-quartz</artifactId><version>2.2.5.RELEASE</version></dependency></dependencies><build><plugins><!--打包跳过测试--><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

resources目录下文件/application.yml:

spring:

profiles:

active: dev

resources目录下文件/application-dev.yml:

server:
port: 12105
servlet:
context-path: /automation-quartz

management:
endpoints:
web:
exposure:
include: '*'

# Spring配置
spring:
resources:
static-locations: classpath:/static/,classpath:/templates/
mvc:
throw-exception-if-no-handler-found: true
static-path-pattern: /**
application:
name: automation-workflow
main:
allow-bean-definition-overriding: true
# 文件上传
servlet:
multipart:
# 单个文件大小
max-file-size: 2000MB
# 设置总上传的文件大小
max-request-size: 4000MB
#json 时间戳统一转换
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
aop:
proxy-target-class: true
autoconfigure:
exclude: com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceAutoConfigure
datasource:
dynamic:
druid:
# 全局druid参数，绝大部分值和默认保持一致。(现已支持的参数如下,不清楚含义不要乱设置)
# 连接池的配置信息
# 初始化大小，最小，最大
initial-size: 1
min-idle: 1
maxActive: 20
# 配置获取连接等待超时的时间
maxWait: 60000
# 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
timeBetweenEvictionRunsMillis: 60000
# 配置一个连接在池中最小生存的时间，单位是毫秒
minEvictableIdleTimeMillis: 300000
validationQuery: SELECT 1
testWhileIdle: true
testOnBorrow: false
testOnReturn: false
# 打开PSCache，并且指定每个连接上PSCache的大小
poolPreparedStatements: true
maxPoolPreparedStatementPerConnectionSize: 20
# 配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
filters: stat,wall,slf4j
# 通过connectProperties属性来打开mergeSql功能；慢SQL记录
connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
datasource:
master:
url: jdbc:mysql://127.0.0.1:3308/merak_dev?characterEncoding=UTF-8&useUnicode=true&useSSL=false
username: root
password: root
driver-class-name: com.mysql.jdbc.Driver

#mybatis plus 设置
mybatis-plus:
mapper-locations: classpath*:com/merak/hyper/automation/persist/**/xml/*Mapper.xml
global-config:
# 关闭MP3.0自带的banner
banner: false
db-config:
id-type: ID_WORKER_STR
# 默认数据库表下划线命名
table-underline: true
configuration:
log-impl: org.apache.ibatis.logging.slf4j.Slf4jImpl
# log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
logging:
level:
com.merar.hyper: debug
com.merak.hyper.automation.persist.**.mapper: debug
org.springframework: warn

2.代码流程

启动MerakQuartzApplication类

packagecom.merak.hyper.automation;importorg.mybatis.spring.annotation.MapperScan;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;importorg.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration;importorg.springframework.context.annotation.Bean;importorg.springframework.scheduling.annotation.EnableAsync;importorg.springframework.scheduling.annotation.EnableScheduling;importorg.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;importorg.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;importjava.util.concurrent.Executor;importjava.util.concurrent.ThreadPoolExecutor;/***@authorchenjun*@version1.0*@ClassName:MerakQuartzApplication*@description:工单任务调度*@date2022/9/2210:30*/@EnableScheduling@EnableAsync@MapperScan(basePackages={"com.merak.hyper.automation.persist.**.mapper"})@SpringBootApplication(scanBasePackages={"com.merak.hyper.automation.**"},exclude={SecurityAutoConfiguration.class})publicclassMerakQuartzApplication{publicstaticfinalLoggerlog=LoggerFactory.getLogger(MerakQuartzApplication.class);publicstaticvoidmain(String[]args){SpringApplication.run(MerakQuartzApplication.class,args);}privateinttaskSchedulerCorePoolSize=15;privateintawaitTerminationSeconds=60;privateStringthreadNamePrefix="taskExecutor-";/***@description:实例化ThreadPoolTaskScheduler对象，用于创建ScheduledFuture<?>scheduledFuture*/@BeanpublicThreadPoolTaskSchedulerthreadPoolTaskScheduler(){ThreadPoolTaskSchedulertaskScheduler=newThreadPoolTaskScheduler();taskScheduler.setPoolSize(taskSchedulerCorePoolSize);taskScheduler.setThreadNamePrefix(threadNamePrefix);taskScheduler.setWaitForTasksToCompleteOnShutdown(false);taskScheduler.setAwaitTerminationSeconds(awaitTerminationSeconds);/**需要实例化线程*/taskScheduler.initialize();//isinitialized=true;log.info("初始化ThreadPoolTaskSchedulerThreadNamePrefix="+threadNamePrefix+",PoolSize="+taskSchedulerCorePoolSize+",awaitTerminationSeconds="+awaitTerminationSeconds);returntaskScheduler;}/***@description:实例化ThreadPoolTaskExecutor对象，管理asyncTask启动的线程，应用类为ScheduledHelper*/@Bean("asyncTaskExecutor")publicExecutortaskExecutor(){ThreadPoolTaskExecutortaskExecutor=newThreadPoolTaskExecutor();taskExecutor.setCorePoolSize(5);taskExecutor.setMaxPoolSize(50);taskExecutor.setQueueCapacity(200);taskExecutor.setKeepAliveSeconds(60);taskExecutor.setThreadNamePrefix("asyncTaskExecutor-");taskExecutor.setWaitForTasksToCompleteOnShutdown(true);taskExecutor.setAwaitTerminationSeconds(60);//修改拒绝策略为使用当前线程执行taskExecutor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());//初始化线程池taskExecutor.initialize();returntaskExecutor;}}

一、启动时项目启动时，加载任务关联的触发器，并全量执行流程。

initLineRunner类：

packagecom.merak.hyper.automation.Scheduling;importcom.baomidou.mybatisplus.core.conditions.query.QueryWrapper;importcom.merak.hyper.automation.persist.entity.AutoTriggerInfo;importcom.merak.hyper.automation.persist.entity.BusWorkflow;importcom.merak.hyper.automation.persist.service.IAutoTriggerInfoService;importcom.merak.hyper.automation.persist.service.IBusWorkflowService;importcom.merak.hyper.automation.util.CommonUtil;importcom.merak.hyper.automation.util.ScheduleUtil;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.CommandLineRunner;importorg.springframework.core.annotation.Order;importorg.springframework.stereotype.Component;importjava.time.LocalDateTime;importjava.time.format.DateTimeFormatter;importjava.util.Iterator;importjava.util.List;importjava.util.Map;/***项目启动时，加载数字员工关联的触发器，并全量执行*@Date:2020/12/25:16:00**/@Component@Order(1)publicclassinitLineRunnerimplementsCommandLineRunner{publicstaticfinalLoggerlog=LoggerFactory.getLogger(initLineRunner.class);privateDateTimeFormatterdf=DateTimeFormatter.ofPattern("yyyy-MM-ddHH:mm:ss");@AutowiredprivateTaskServicetaskService;@AutowiredprivateIAutoTriggerInfoServicetriggerInfoService;@AutowiredprivateIBusWorkflowServiceworkflowService;@Overridepublicvoidrun(String...args){log.info("项目启动:加载数字员工关联的触发器信息并全量执行,"+df.format(LocalDateTime.now()));QueryWrapper<BusWorkflow>wrapper=newQueryWrapper<>();wrapper.eq("wf_type","3");//3:云托管wrapper.eq("wf_state","1");List<BusWorkflow>busWorkflows=workflowService.list(wrapper);List<AutoTriggerInfo>triggerInfos=triggerInfoService.list();if(0==busWorkflows.size()||0==triggerInfos.size()){log.info("数字员工关联的触发器信息不正确，员工记录数:"+busWorkflows.size()+",触发器记录数:"+triggerInfos.size());}else{//数字员工关联的触发器信息Map<String,AutoTriggerInfo>loadWfidAndTriggerInfo=CommonUtil.loadWfidAndTriggerInfo(busWorkflows,triggerInfos);Iterator<Map.Entry<String,AutoTriggerInfo>>entries=loadWfidAndTriggerInfo.entrySet().iterator();while(entries.hasNext()){Map.Entry<String,AutoTriggerInfo>entry=entries.next();StringwfId=entry.getKey();BusWorkflowworkflow=busWorkflows.stream().filter(t->wfId.equals(t.getWfId())).findAny().orElse(null);if(null!=workflow){ScheduleUtil.start(newScheduleTask(wfId,String.valueOf(workflow.getWfCreateuserId()),taskService),entry.getValue());}}log.info("数字员工关联的触发器信息全量执行完成，数字员工定时个数:"+loadWfidAndTriggerInfo.size()+","+df.format(LocalDateTime.now()));}}}核心代码：```javaScheduleUtil.start(newScheduleTask(wfId,String.valueOf(workflow.getWfCreateuserId()),taskService),entry.getValue());

Scheduler管理工具类:启动、取消、修改等管理

packagecom.merak.hyper.automation.util;importcom.merak.hyper.automation.Scheduling.ScheduleTask;importcom.merak.hyper.automation.persist.entity.AutoTriggerInfo;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;importorg.springframework.scheduling.support.CronTrigger;importjava.util.HashMap;importjava.util.Map;importjava.util.concurrent.ScheduledFuture;/***@version1.0*@ClassName:ScheduleUtil*@description:Scheduler管理工具类:启动、取消、修改等管理*/publicclassScheduleUtil{publicstaticfinalLoggerlog=LoggerFactory.getLogger(ScheduleUtil.class);privatestaticThreadPoolTaskSchedulerthreadPoolTaskScheduler=SpringContextUtils.getBean(ThreadPoolTaskScheduler.class);//存储[数字员工wfI，dScheduledFuture]集合privatestaticMap<String,ScheduledFuture<?>>scheduledFutureMap=newHashMap<>();/***启动**@paramscheduleTask定时任务*@paramtriggerInfo*/publicstaticbooleanstart(ScheduleTaskscheduleTask,AutoTriggerInfotriggerInfo){StringwfId=scheduleTask.getId();log.info("启动数字员工"+wfId+"定时任务线程"+scheduleTask.getId());ScheduledFuture<?>scheduledFuture=threadPoolTaskScheduler.schedule(scheduleTask,newCronTrigger(triggerInfo.getLogicConfig()));scheduledFutureMap.put(wfId,scheduledFuture);returntrue;}/***取消**@paramscheduleTask定时任务*/publicstaticbooleancancel(ScheduleTaskscheduleTask){log.info("关闭定时任务线程taskId"+scheduleTask.getId());ScheduledFuture<?>scheduledFuture=scheduledFutureMap.get(scheduleTask.getId());if(scheduledFuture!=null&&!scheduledFuture.isCancelled()){scheduledFuture.cancel(false);}scheduledFutureMap.remove(scheduleTask.getId());returntrue;}/***修改**@paramscheduleTask定时任务*@paramtriggerInfo*/publicstaticbooleanreset(ScheduleTaskscheduleTask,AutoTriggerInfotriggerInfo){//先取消定时任务cancel(scheduleTask);//然后启动新的定时任务start(scheduleTask,triggerInfo);returntrue;}}

ScheduleTask类：ScheduleTask任务类

packagecom.merak.hyper.automation.Scheduling;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;/***@version1.0*@ClassName:ScheduleTask*@description:ScheduleTask，关联任务id、用户id和具体执行的TaskService类，实现Runnable类*/publicclassScheduleTaskimplementsRunnable{privatestaticfinalintTIMEOUT=30000;privateStringid;privateStringuserId;privateTaskServiceservice;publicstaticfinalLoggerlog=LoggerFactory.getLogger(ScheduleTask.class);publicStringgetId(){returnid;}/***@paramid任务ID*@paramservice业务类*/publicScheduleTask(Stringid,StringuserId,TaskServiceservice){this.id=id;this.userId=userId;this.service=service;}@Overridepublicvoidrun(){log.info("ScheduleTask-执行数字员工消息的发送,id:"+this.id+",用户id:"+userId);service.work(this.id,this.userId);}}

/***@version1.0*@ClassName:TaskService*@description:TaskService*/publicinterfaceTaskService{/***业务处理方法*@paramkeyword关键参数*@paramuserId*/voidwork(Stringkeyword,StringuserId);}/***@description:TaskService实现类，具体执行定时调度的业务*/@ServicepublicclassTaskServiceImplimplementsTaskService{publicstaticfinalLoggerlog=LoggerFactory.getLogger(TaskServiceImpl.class);@AutowiredprivateIAutoDeviceInfoServicedeviceInfoService;@Overridepublicvoidwork(StringwfId,StringuserId){try{log.info("定时任务：根据数字员工wfId"+wfId+",用户id:"+userId+"，发送消息...");//sendRobotMsg(wfId,userId);Thread.sleep(200);}catch(InterruptedExceptione){e.printStackTrace();}}

二、通过WEB配置的变更，动态管理定时任务

ScheduledController类:scheduled Web业务层：启动、取消、修改等管理schedule

调度任务信息变更（如1：Trigger Cron变更 2：任务停止 3：任务新增加等）

packagecom.merak.hyper.automation.controller;importcom.merak.hyper.automation.common.core.domain.AjaxResult;importcom.merak.hyper.automation.common.core.vo.ScheduledApiVo;importcom.merak.hyper.automation.persist.entity.AutoTriggerInfo;importcom.merak.hyper.automation.persist.service.IAutoTriggerInfoService;importcom.merak.hyper.automation.util.ScheduledHelper;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.*;/***@version1.0*@ClassName:ScheduledController*@description:scheduledWeb业务层：启动、取消、修改等管理schedule*/@RestController@RequestMapping("/api/scheduled")publicclassScheduledController{publicstaticfinalLoggerlog=LoggerFactory.getLogger(ScheduledController.class);@AutowiredprivateIAutoTriggerInfoServicetriggerInfoService;@AutowiredprivateScheduledHelperscheduledHelper;@PostMapping("/add")publicAjaxResultaddScheduleds(@RequestBodyScheduledApiVoscheduledApiVo){AutoTriggerInfoautoTriggerInfo=triggerInfoService.getById(scheduledApiVo.getTriggerId());scheduledHelper.addScheduleds(scheduledApiVo,autoTriggerInfo);returnAjaxResult.success();}@PostMapping("/reset")publicAjaxResultresetScheduleds(@RequestBodyScheduledApiVoscheduledApiVo){AutoTriggerInfoautoTriggerInfo=triggerInfoService.getById(scheduledApiVo.getTriggerId());scheduledHelper.resetScheduleds(scheduledApiVo,autoTriggerInfo);returnAjaxResult.success();}@PostMapping("/stop")publicAjaxResultstopScheduleds(@RequestBodyScheduledApiVoscheduledApiVo){AutoTriggerInfoautoTriggerInfo=triggerInfoService.getById(scheduledApiVo.getTriggerId());scheduledHelper.stopScheduleds(scheduledApiVo);returnAjaxResult.success();}}ScheduledHelper类：对外提供ScheduledHelper管理:创建、变更、停止```javapackagecom.merak.hyper.automation.util;importcom.merak.hyper.automation.Scheduling.ScheduleTask;importcom.merak.hyper.automation.Scheduling.TaskService;importcom.merak.hyper.automation.common.core.vo.ScheduledApiVo;importcom.merak.hyper.automation.persist.entity.AutoTriggerInfo;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.scheduling.annotation.Async;importorg.springframework.stereotype.Component;/***@version1.0*@ClassName:ScheduledHelper*@description:对外提供ScheduledHelper管理:创建、变更、停止*/@ComponentpublicclassScheduledHelper{publicstaticfinalLoggerlog=LoggerFactory.getLogger(ScheduledHelper.class);/***@description:对外（Web)提供异步的Scheduleds增加操作*/@Async("asyncTaskExecutor")publicvoidaddScheduleds(ScheduledApiVoscheduledApiVo,AutoTriggerInfotriggerInfo){//addSchedule任务log.warn("创建原数字员工["+scheduledApiVo.getWfId()+"],同步启动Schedule任务");TaskServicetaskService=SpringContextUtils.getBean(TaskService.class);ScheduleUtil.start(newScheduleTask(scheduledApiVo.getWfId(),scheduledApiVo.getUserId(),taskService),triggerInfo);}@Async("asyncTaskExecutor")publicvoidresetScheduleds(ScheduledApiVoscheduledApiVo,AutoTriggerInfotriggerInfo){//cron值改变，变更Schedule任务log.warn("数字员工["+scheduledApiVo.getWfId()+"]关联的触发器信息cron值改变，变更Schedule任务");TaskServicetaskService=SpringContextUtils.getBean(TaskService.class);ScheduleUtil.reset(newScheduleTask(scheduledApiVo.getWfId(),scheduledApiVo.getUserId(),taskService),triggerInfo);}@Async("asyncTaskExecutor")publicvoidstopScheduleds(ScheduledApiVoscheduledApiVo){//移除Wfid，停止原Schedule任务log.warn("原数字员工["+scheduledApiVo.getWfId()+"]无效，同步停止Schedule任务");TaskServicetaskService=SpringContextUtils.getBean(TaskService.class);ScheduleUtil.cancel(newScheduleTask(scheduledApiVo.getWfId(),scheduledApiVo.getUserId(),taskService));}}

SpringContextUtils类：

packagecom.merak.hyper.automation.util;importorg.springframework.beans.BeansException;importorg.springframework.context.ApplicationContext;importorg.springframework.context.ApplicationContextAware;importorg.springframework.stereotype.Component;/***@version1.0*@ClassName:SpringContextUtils*@description:加载Class对象*/@ComponentpublicclassSpringContextUtilsimplementsApplicationContextAware{privatestaticApplicationContextapplicationContext;@OverridepublicvoidsetApplicationContext(ApplicationContextapplicationContext)throwsBeansException{SpringContextUtils.applicationContext=applicationContext;}publicstaticObjectgetBean(Stringname){returnapplicationContext.getBean(name);}publicstatic<T>TgetBean(Class<T>requiredType){returnapplicationContext.getBean(requiredType);}publicstatic<T>TgetBean(Stringname,Class<T>requiredType){returnapplicationContext.getBean(name,requiredType);}publicstaticbooleancontainsBean(Stringname){returnapplicationContext.containsBean(name);}publicstaticbooleanisSingleton(Stringname){returnapplicationContext.isSingleton(name);}publicstaticClass<?extendsObject>getType(Stringname){returnapplicationContext.getType(name);}}

ScheduledApiVo类：

importjava.io.Serializable;/***@version1.0*@ClassName:ScheduledApiVo*@description:scheduledWeb业务层Api传递参数Vo类*/publicclassScheduledApiVoimplementsSerializable{privateStringwfId;privateStringuserId;privateStringtriggerId;//setget略}

最终：Web端通过发送Http请求 ，调用ScheduledHelper管理类接口，实现Scheduled创建、变更、停止操作

log.info("3:云托管更新启动数字员工操作");ScheduledApiVoscheduledApiVo=newScheduledApiVo();scheduledApiVo.setWfId(wfId);scheduledApiVo.setUserId(String.valueOf(updateUserId));scheduledApiVo.setTriggerId(newTriggerInfo.getId());StringwebHookBody=JSON.toJSONString(scheduledApiVo);EmsApiUtil.SendQuartzMessage(url,"add",webHookBody);********************分隔************************publicstaticbooleanSendQuartzMessage(StringquartzUrl,Stringmethod,StringwebHookBody){booleanresult=false;try{//org.apache.httpcomponents.httpclientsendPost，pom依赖如下dependencyStringresp=HttpClientUtil.sendPostByJson(quartzUrl+"/"+method,webHookBody,0);if("error".equals(resp)||resp.contains("405NotAllowed")){log.error("调用任务调度中心消息发送失败,地址:"+quartzUrl);}else{JSONObjectjsonObject=JSON.parseObject(resp);if("200".equals(String.valueOf(jsonObject.get("code")))){result=true;}else{log.error("调用任务调度中心失败,msg："+String.valueOf(jsonObject.get("msg")));}}}catch(Exceptione){log.error("调用任务调度中心失败,msg："+e.getMessage());}returnresult;}

<dependency><groupId>org.apache.httpcomponents</groupId><artifactId>httpclient</artifactId><version>4.5.2</version></dependency>