定时任务动态管理分为两种方式:
方式一:Web前台配置Trigger触发器(关联Cron)、ThreadPoolTaskScheduler类创建Scheduler方式下进行Schedule调度任务的动态管理
方式二:基于已创建的Schedule调度任务的动态管理,即以组件类 @Scheduled注解声明Schedule调度,在启动程序前一次性初始化,如:
@Component public class TestTask { private DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); @Scheduled(cron = "0/2 * * * * ?") public void robReceiveExpireTask() { System.out.println(df.format(LocalDateTime.now()) + "测试测试"); } }
缺陷:目前无法在运行期间增加Schedule以及stop、Start、Reset等管理。
架构为SpringBoot + Spring + mybatis-plus
pom.xml
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="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.0 http://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>aliyun Repository</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-quartzmanagement:
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
启动MerakQuartzApplication类
package com.merak.hyper.automation; import org.mybatis.spring.annotation.MapperScan; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import java.util.concurrent.Executor; import java.util.concurrent.ThreadPoolExecutor; /** * @author chenjun * @version 1.0 * @ClassName: MerakQuartzApplication * @description: 工单任务调度 * @date 2022/9/22 10:30 */ @EnableScheduling @EnableAsync @MapperScan(basePackages = {"com.merak.hyper.automation.persist.**.mapper"}) @SpringBootApplication(scanBasePackages = {"com.merak.hyper.automation.**"}, exclude = {SecurityAutoConfiguration.class}) public class MerakQuartzApplication { public static final Logger log = LoggerFactory.getLogger(MerakQuartzApplication.class); public static void main(String[] args) { SpringApplication.run(MerakQuartzApplication.class, args); } private int taskSchedulerCorePoolSize = 15; private int awaitTerminationSeconds = 60; private String threadNamePrefix = "taskExecutor-"; /** * @description: 实例化ThreadPoolTaskScheduler对象,用于创建ScheduledFuture<?> scheduledFuture */ @Bean public ThreadPoolTaskScheduler threadPoolTaskScheduler() { ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.setPoolSize(taskSchedulerCorePoolSize); taskScheduler.setThreadNamePrefix(threadNamePrefix); taskScheduler.setWaitForTasksToCompleteOnShutdown(false); taskScheduler.setAwaitTerminationSeconds(awaitTerminationSeconds); /**需要实例化线程*/ taskScheduler.initialize(); // isinitialized = true; log.info("初始化ThreadPoolTaskScheduler ThreadNamePrefix=" + threadNamePrefix + ",PoolSize=" + taskSchedulerCorePoolSize + ",awaitTerminationSeconds=" + awaitTerminationSeconds); return taskScheduler; } /** * @description: 实例化ThreadPoolTaskExecutor对象,管理asyncTask启动的线程,应用类为 ScheduledHelper */ @Bean("asyncTaskExecutor") public Executor taskExecutor() { ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); taskExecutor.setCorePoolSize(5); taskExecutor.setMaxPoolSize(50); taskExecutor.setQueueCapacity(200); taskExecutor.setKeepAliveSeconds(60); taskExecutor.setThreadNamePrefix("asyncTaskExecutor-"); taskExecutor.setWaitForTasksToCompleteOnShutdown(true); taskExecutor.setAwaitTerminationSeconds(60); //修改拒绝策略为使用当前线程执行 taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); //初始化线程池 taskExecutor.initialize(); return taskExecutor; } }
一、启动时项目启动时,加载任务关联的触发器,并全量执行流程。
initLineRunner类:
package com.merak.hyper.automation.Scheduling; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.merak.hyper.automation.persist.entity.AutoTriggerInfo; import com.merak.hyper.automation.persist.entity.BusWorkflow; import com.merak.hyper.automation.persist.service.IAutoTriggerInfoService; import com.merak.hyper.automation.persist.service.IBusWorkflowService; import com.merak.hyper.automation.util.CommonUtil; import com.merak.hyper.automation.util.ScheduleUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.CommandLineRunner; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.Iterator; import java.util.List; import java.util.Map; /** * 项目启动时,加载数字员工关联的触发器,并全量执行 * @Date: 2020/12/25:16:00 **/ @Component @Order(1) public class initLineRunner implements CommandLineRunner { public static final Logger log = LoggerFactory.getLogger(initLineRunner.class); private DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); @Autowired private TaskService taskService; @Autowired private IAutoTriggerInfoService triggerInfoService; @Autowired private IBusWorkflowService workflowService; @Override public void run(String... args) { log.info("项目启动:加载数字员工关联的触发器信息并全量执行," + df.format(LocalDateTime.now())); QueryWrapper<BusWorkflow> wrapper = new QueryWrapper<>(); 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(); String wfId = entry.getKey(); BusWorkflow workflow = busWorkflows.stream().filter( t -> wfId.equals(t.getWfId()) ).findAny().orElse(null); if( null != workflow ){ ScheduleUtil.start(new ScheduleTask(wfId,String.valueOf(workflow.getWfCreateuserId()),taskService), entry.getValue()); } } log.info("数字员工关联的触发器信息全量执行完成,数字员工定时个数:"+loadWfidAndTriggerInfo.size()+","+df.format(LocalDateTime.now())); } } } 核心代码: ```java ScheduleUtil.start(new ScheduleTask(wfId,String.valueOf(workflow.getWfCreateuserId()),taskService), entry.getValue());
Scheduler管理工具类:启动、取消、修改等管理
package com.merak.hyper.automation.util; import com.merak.hyper.automation.Scheduling.ScheduleTask; import com.merak.hyper.automation.persist.entity.AutoTriggerInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.support.CronTrigger; import java.util.HashMap; import java.util.Map; import java.util.concurrent.ScheduledFuture; /** * @version 1.0 * @ClassName: ScheduleUtil * @description: Scheduler管理工具类:启动、取消、修改等管理 */ public class ScheduleUtil { public static final Logger log = LoggerFactory.getLogger(ScheduleUtil.class); private static ThreadPoolTaskScheduler threadPoolTaskScheduler = SpringContextUtils.getBean(ThreadPoolTaskScheduler.class); //存储[数字员工wfI,dScheduledFuture]集合 private static Map<String, ScheduledFuture<?>> scheduledFutureMap = new HashMap<>(); /** * 启动 * * @param scheduleTask 定时任务 * @param triggerInfo */ public static boolean start(ScheduleTask scheduleTask, AutoTriggerInfo triggerInfo) { String wfId = scheduleTask.getId(); log.info("启动数字员工"+wfId+"定时任务线程" + scheduleTask.getId()); ScheduledFuture<?> scheduledFuture = threadPoolTaskScheduler.schedule(scheduleTask, new CronTrigger(triggerInfo.getLogicConfig())); scheduledFutureMap.put(wfId, scheduledFuture); return true; } /** * 取消 * * @param scheduleTask 定时任务 */ public static boolean cancel(ScheduleTask scheduleTask) { log.info("关闭定时任务线程 taskId " + scheduleTask.getId()); ScheduledFuture<?> scheduledFuture = scheduledFutureMap.get(scheduleTask.getId()); if (scheduledFuture != null && !scheduledFuture.isCancelled()) { scheduledFuture.cancel(false); } scheduledFutureMap.remove(scheduleTask.getId()); return true; } /** * 修改 * * @param scheduleTask 定时任务 * @param triggerInfo */ public static boolean reset(ScheduleTask scheduleTask, AutoTriggerInfo triggerInfo) { //先取消定时任务 cancel(scheduleTask); //然后启动新的定时任务 start(scheduleTask, triggerInfo); return true; } }
ScheduleTask类:ScheduleTask任务类
package com.merak.hyper.automation.Scheduling; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * @version 1.0 * @ClassName: ScheduleTask * @description: ScheduleTask,关联任务id、用户id和具体执行的TaskService类,实现Runnable类 */ public class ScheduleTask implements Runnable { private static final int TIMEOUT = 30000; private String id; private String userId; private TaskService service; public static final Logger log = LoggerFactory.getLogger(ScheduleTask.class); public String getId() { return id; } /** * @param id 任务ID * @param service 业务类 */ public ScheduleTask(String id, String userId, TaskService service) { this.id = id; this.userId = userId; this.service = service; } @Override public void run() { log.info("ScheduleTask-执行数字员工消息的发送,id:"+ this.id + ",用户id:"+userId); service.work(this.id,this.userId); } }
/** * @version 1.0 * @ClassName: TaskService * @description: TaskService */ public interface TaskService { /** * 业务处理方法 * @param keyword 关键参数 * @param userId */ void work(String keyword,String userId); } /** * @description: TaskService实现类,具体执行定时调度的业务 */ @Service public class TaskServiceImpl implements TaskService { public static final Logger log = LoggerFactory.getLogger(TaskServiceImpl.class); @Autowired private IAutoDeviceInfoService deviceInfoService; @Override public void work(String wfId,String userId) { try { log.info("定时任务:根据数字员工wfId"+ wfId +",用户id:"+userId+",发送消息..."); //sendRobotMsg(wfId,userId); Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); } }
二、通过WEB配置的变更,动态管理定时任务
ScheduledController类:scheduled Web业务层:启动、取消、修改等管理schedule
调度任务信息变更(如1:Trigger Cron变更 2:任务停止 3:任务新增加等)
package com.merak.hyper.automation.controller; import com.merak.hyper.automation.common.core.domain.AjaxResult; import com.merak.hyper.automation.common.core.vo.ScheduledApiVo; import com.merak.hyper.automation.persist.entity.AutoTriggerInfo; import com.merak.hyper.automation.persist.service.IAutoTriggerInfoService; import com.merak.hyper.automation.util.ScheduledHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.*; /** * @version 1.0 * @ClassName: ScheduledController * @description: scheduled Web业务层:启动、取消、修改等管理schedule */ @RestController @RequestMapping("/api/scheduled") public class ScheduledController { public static final Logger log = LoggerFactory.getLogger(ScheduledController.class); @Autowired private IAutoTriggerInfoService triggerInfoService; @Autowired private ScheduledHelper scheduledHelper; @PostMapping("/add") public AjaxResult addScheduleds(@RequestBody ScheduledApiVo scheduledApiVo){ AutoTriggerInfo autoTriggerInfo = triggerInfoService.getById(scheduledApiVo.getTriggerId()); scheduledHelper.addScheduleds(scheduledApiVo,autoTriggerInfo); return AjaxResult.success(); } @PostMapping("/reset") public AjaxResult resetScheduleds(@RequestBody ScheduledApiVo scheduledApiVo){ AutoTriggerInfo autoTriggerInfo = triggerInfoService.getById(scheduledApiVo.getTriggerId()); scheduledHelper.resetScheduleds(scheduledApiVo,autoTriggerInfo); return AjaxResult.success(); } @PostMapping("/stop") public AjaxResult stopScheduleds(@RequestBody ScheduledApiVo scheduledApiVo){ AutoTriggerInfo autoTriggerInfo = triggerInfoService.getById(scheduledApiVo.getTriggerId()); scheduledHelper.stopScheduleds(scheduledApiVo); return AjaxResult.success(); } } ScheduledHelper类:对外提供ScheduledHelper管理:创建、变更、停止 ```java package com.merak.hyper.automation.util; import com.merak.hyper.automation.Scheduling.ScheduleTask; import com.merak.hyper.automation.Scheduling.TaskService; import com.merak.hyper.automation.common.core.vo.ScheduledApiVo; import com.merak.hyper.automation.persist.entity.AutoTriggerInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Component; /** * @version 1.0 * @ClassName: ScheduledHelper * @description:对外提供ScheduledHelper管理:创建、变更、停止 */ @Component public class ScheduledHelper { public static final Logger log = LoggerFactory.getLogger(ScheduledHelper.class); /** * @description: 对外(Web)提供异步的Scheduleds增加操作 */ @Async("asyncTaskExecutor") public void addScheduleds(ScheduledApiVo scheduledApiVo, AutoTriggerInfo triggerInfo) { //addSchedule任务 log.warn("创建原数字员工["+scheduledApiVo.getWfId()+"],同步启动Schedule任务"); TaskService taskService = SpringContextUtils.getBean(TaskService.class); ScheduleUtil.start(new ScheduleTask(scheduledApiVo.getWfId(), scheduledApiVo.getUserId(), taskService), triggerInfo); } @Async("asyncTaskExecutor") public void resetScheduleds(ScheduledApiVo scheduledApiVo,AutoTriggerInfo triggerInfo) { //cron值改变,变更Schedule任务 log.warn("数字员工["+scheduledApiVo.getWfId()+"]关联的触发器信息cron值改变,变更Schedule任务"); TaskService taskService = SpringContextUtils.getBean(TaskService.class); ScheduleUtil.reset(new ScheduleTask(scheduledApiVo.getWfId(), scheduledApiVo.getUserId(), taskService), triggerInfo); } @Async("asyncTaskExecutor") public void stopScheduleds(ScheduledApiVo scheduledApiVo) { //移除Wfid,停止原Schedule任务 log.warn("原数字员工["+scheduledApiVo.getWfId()+"]无效,同步停止Schedule任务"); TaskService taskService = SpringContextUtils.getBean(TaskService.class); ScheduleUtil.cancel(new ScheduleTask(scheduledApiVo.getWfId(), scheduledApiVo.getUserId(), taskService)); } }
SpringContextUtils类:
package com.merak.hyper.automation.util; import org.springframework.beans.BeansException; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.stereotype.Component; /** * @version 1.0 * @ClassName: SpringContextUtils * @description: 加载Class对象 */ @Component public class SpringContextUtils implements ApplicationContextAware { private static ApplicationContext applicationContext; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { SpringContextUtils.applicationContext = applicationContext; } public static Object getBean(String name) { return applicationContext.getBean(name); } public static <T> T getBean(Class<T> requiredType) { return applicationContext.getBean(requiredType); } public static <T> T getBean(String name, Class<T> requiredType) { return applicationContext.getBean(name, requiredType); } public static boolean containsBean(String name) { return applicationContext.containsBean(name); } public static boolean isSingleton(String name) { return applicationContext.isSingleton(name); } public static Class<? extends Object> getType(String name) { return applicationContext.getType(name); } }
ScheduledApiVo类:
import java.io.Serializable; /** * @version 1.0 * @ClassName: ScheduledApiVo * @description: scheduled Web业务层Api传递参数Vo类 */ public class ScheduledApiVo implements Serializable { private String wfId; private String userId; private String triggerId; //set get 略 }
最终:Web端通过发送Http请求 ,调用ScheduledHelper管理类接口,实现Scheduled创建、变更、停止操作
log.info("3:云托管更新启动数字员工操作"); ScheduledApiVo scheduledApiVo = new ScheduledApiVo(); scheduledApiVo.setWfId(wfId); scheduledApiVo.setUserId(String.valueOf(updateUserId)); scheduledApiVo.setTriggerId(newTriggerInfo.getId()); String webHookBody = JSON.toJSONString(scheduledApiVo); EmsApiUtil.SendQuartzMessage(url, "add", webHookBody); ******************** 分隔 ************************ public static boolean SendQuartzMessage(String quartzUrl, String method, String webHookBody){ boolean result = false; try{ //org.apache.httpcomponents.httpclient sendPost,pom依赖如下dependency String resp = HttpClientUtil.sendPostByJson(quartzUrl+"/"+method, webHookBody,0); if( "error".equals(resp) || resp.contains("405 Not Allowed")){ log.error("调用任务调度中心消息发送失败,地址:"+quartzUrl); } else { JSONObject jsonObject = JSON.parseObject(resp); if( "200".equals(String.valueOf(jsonObject.get("code"))) ){ result = true; } else{ log.error("调用任务调度中心失败,msg:"+String.valueOf(jsonObject.get("msg"))); } } }catch (Exception e){ log.error("调用任务调度中心失败,msg:"+e.getMessage()); } return result; }
<dependency> <groupId>org.apache.httpcomponents</groupId> <artifactId>httpclient</artifactId> <version>4.5.2</version> </dependency>
以上是SpringBoot Schedule调度任务的动态管理方法是什么的详细内容。更多信息请关注PHP中文网其他相关文章!