关于对Spring定时任务的使用

2022-12-04 05:32| 来源: 网络整理| 查看: 265

为了完成既能通过数据库修改定时任务，也能通过接口实现定时任务的功能

而且不用框架，这就要用到Spring的定时任务线程池了，

首先创建一个类创建线程池

public class OrderCollectScheduledConfig { /** * 设置线程池，防止多个任务同步执行造成部分数据不会显示 * @return */ @Bean public TaskScheduler taskScheduler() { log.info("创建定时任务调度线程池 start"); ThreadPoolTaskScheduler executor = new ThreadPoolTaskScheduler(); executor.setPoolSize(10); executor.setThreadNamePrefix("taskExecutor-"); executor.setRemoveOnCancelPolicy(true); //设置饱和策略 //CallerRunsPolicy：线程池的饱和策略之一，当线程池使用饱和后，直接使用调用者所在的线程来执行任务；如果执行程序已关闭，则会丢弃该任务 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.initialize(); log.info("创建定时任务调度线程池 end"); return executor; } }

然后生成一个实现了Runable接口的类

如下

public class OrderCollectRunable implements Runnable{ private ScmJob scmJob; public OrderCollectRunable(ScmJob scmJob) { this.scmJob=scmJob; } @Override public void run() { try { Class clazz = Class.forName(scmJob.getBeanName()); Object bean = SpringContextUtils.getBean(clazz); Method method = ReflectionUtils.findMethod(bean.getClass(),scmJob.getMethodName(),ScmJob.class); ReflectionUtils.invokeMethod(method,bean,scmJob); } catch (ClassNotFoundException e) { throw new RuntimeException(e); } } @Override public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; OrderCollectRunable that = (OrderCollectRunable) o; return scmJob.equals(that.scmJob); } @Override public int hashCode() { return Objects.hash(scmJob); } }

其中Run方法是你需要的业务逻辑，

然后来到核心代码：

public class CronTaskRegistrar { private final Map scheduledFutureMap = new ConcurrentHashMap(16); public final Map map=new HashMap(); @Autowired private TaskScheduler taskScheduler; @Autowired private ThreadPoolTaskScheduler threadPoolTaskScheduler; @Autowired private ScmJobService scmJobService; private static final Logger LOGGER = LoggerFactory.getLogger(CronTaskRegistrar.class); public void start(ScmJob scmJob) { LOGGER.info("准备启动任务：{}", scmJob.getJobName()); try { //校验是否已经启动 if (this.isStart(scmJob)) { LOGGER.info("当前任务已在启动列表，请不要重复启动！"); }else{ //启动任务 this.doStart(scmJob); } }catch (Exception e) { e.printStackTrace(); } } /** * 根据任务id 判断定时任务是否启动 */ public Boolean isStart(ScmJob scmJob) { //首先检查scheduledFutureMap是否存在该任务，如果不存在，则确定当前任务并没有启动 if (scheduledFutureMap.containsKey(scmJob)) { //当该任务存在时，需要检查scheduledFuture对象是否被取消，如果为false，说明当前线程在启动，否则当前线程处于关闭状态 if (!scheduledFutureMap.get(scmJob).future.isCancelled()) { return true; } } return false; } /** * 根据任务id 停止定时任务 * 该方法加锁，避免 */ public void stop(ScmJob scmJob) { LOGGER.info("进入关闭定时任务：{}", scmJob.getJobName()); //首先检查当前任务实例是否存在 if (scheduledFutureMap.containsKey(scmJob)) { try { //获取任务实例 ScheduledTask scheduledTask = scheduledFutureMap.get(scmJob); //关闭定时任务 scheduledTask.cancel(); //避免内存泄露 //scheduledFutureMap.remove(id); LOGGER.info("任务{}已成功关闭", scmJob.getJobName()); }catch (Exception e) { e.printStackTrace(); } }else { LOGGER.info("当前任务{}不存在，请重试！", scmJob.getJobName()); } } public void init(ScmJob scmJob){ LOGGER.info("定时任务开始初始化:"+scmJob.getJobName()); //如果集合为空，则直接退出当前方法 if (scmJob==null) { return; } else { if (!isStart(scmJob)){ doStart(scmJob); } } } /** * 启动定时任务(该方法设置为私有方法，不开放给对象直接调用) */ private void doStart(ScmJob scmJob) { OrderCollectRunable orderCollectRunable = new OrderCollectRunable(scmJob); ScheduledTask scheduledTask = new ScheduledTask(); ScheduledFuture scheduledFuture = threadPoolTaskScheduler.schedule(orderCollectRunable, new Trigger() { @Override public Date nextExecutionTime(TriggerContext triggerContext) { CronTrigger cronTrigger = new CronTrigger(scmJob.getCron()); return cronTrigger.nextExecutionTime(triggerContext); } }); scheduledTask.future=scheduledFuture; //将已经启动的定时任务实例放入scheduledFutureMap进行统一管理 scheduledFutureMap.put(scmJob, scheduledTask); map.put(scmJob.getJobName(),scmJob); LOGGER.info("启动任务：{} 成功！",scmJob.getJobName()); } }

通过对threadPoolTaskScheduler.schedule的操作将定时任务放到线程池中，这样就可以手动的去取消或者增加定时任务

注意这里要重写实体类的equals方法和Hashcode方法

要不然比对不上，还有map的作用是为了存储老的定时任务的实体类，方便下次做比对，下次的实体类和这次的实体类相同，则不修改定时任务

接下来就是细枝末节的东西了

public final class ScheduledTask { volatile ScheduledFuture future; /** * 取消定时任务 */ public void cancel() { ScheduledFuture future = this.future; if (future != null) { future.cancel(true); } } }

@Component public class RunConfig implements ApplicationRunner { @Resource private ScmJobService scmJobService; @Resource private CronTaskRegistrar cronTaskRegistrar; @Override public void run(ApplicationArguments args) throws Exception { LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper(); queryWrapper.eq(ScmJob::getJobName,"orderCollect"); ScmJob start = scmJobService.getOne(queryWrapper); if (start!=null){ cronTaskRegistrar.start(start); } //TODO 执行线程同步操作 ScmJob synchronization = new ScmJob(); synchronization.setJobName("synchronization"); synchronization.setCron("0/5 * * * * ?"); synchronization.setBeanName("com.jxaisino.scm.controller.ScmOrderCollectController"); synchronization.setMethodName("synchronizationOrderCollectSetting"); cronTaskRegistrar.start(synchronization); }

这个类是为了项目启动时就加入同步线程

public void synchronizationOrderCollectSetting(ScmJob scmJob) { LambdaQueryWrapper lambdaQueryWrapper = new LambdaQueryWrapper(); lambdaQueryWrapper.eq(ScmJob::getJobName,"orderCollect"); ScmJob newScmJob = this.getOne(lambdaQueryWrapper); CronTaskRegistrar cronTaskRegistrar = SpringContextUtils.getBean(CronTaskRegistrar.class); ScmJob oldScmJob = cronTaskRegistrar.map.get("orderCollect"); if (newScmJob!=null&&!newScmJob.equals(oldScmJob)){ if (oldScmJob!=null){ cronTaskRegistrar.stop(oldScmJob); } cronTaskRegistrar.start(newScmJob); } else if (newScmJob==null&&oldScmJob!=null){ cronTaskRegistrar.stop(oldScmJob); cronTaskRegistrar.map.remove("orderCollect"); } }

最后是线程同步的方法，通过比对新的实体类与老的实体类是否一致而决定如何操作

【本文地址】

公司简介

联系我们