spring schedule配置多任务动态cron(增删启停)

2024-04-02 19:04:59 593人浏览八月长安

Python 官方文档：入门教程 => 点击学习

摘要

一、背景之前公司经常会遇到配置定时任务，简单的任务可以直接依赖spring。简单任务直接使用 @scheduled 注解配合@EnableScheduling。但是如何实现简单

一、背景

之前公司经常会遇到配置定时任务，简单的任务可以直接依赖spring。
简单任务直接使用 @scheduled 注解配合@EnableScheduling。
但是如何实现简单的动态cron呢？

开发原则：
尽可能在项目本身去实现，少依赖第三方框架，避免项目过于臃肿和复杂。

俩种任务调度方式：

在这里插入图片描述

二、本篇说明

SpringBoot 基础模块 spring-boot-starter-WEB 已经内置 schedule ，无需引入额外依赖。
先思考几个问题：

1、动态 cron 实现的原理

任务的【停止】是基于 future接口的cancel() 方法。
任务的【增加、删除、启动】是基于注册到类ScheduledTaskReGIStrar 的 ScheduledFuture的数量。
涉及核心类：

ScheduledFuture
SchedulinGConfigurer
ScheduledTaskRegistrar

2、多任务并行执行配置
spring默认机制对schedule是单线程，需要配置多线程并行执行。

3、如何配置多个任务
好多博文，都是配置一个cron，这让初学者很难受。

4、如何配置任务分组
根据自己业务背景，可根据步骤三，进行改造。

5、如何配置服务启动自启任务。
想要程序启动时首次去加我们设置的task，只需实现 CommandLineRunner 即可。

6、如何从数据库读取配置
这个其实很简单，在实现 ScheduledTaskRegistrar 时，先直接查询我们需要的数据即可。

7、如何优雅的实现我们的代码
这里为了我们多个task实现时，去除臃肿的if else ，使用策略模式去实现我们的task，这里代码里面会具体介绍。

参考类图：

在这里插入图片描述

8、如何去触发我们的schedule 【增删启停】
配置好 task任务类，注入到 controller ，通过接口直接调用即可。

三、代码实现

先贴出我的GitHub 代码，下面代码描述不全。

普通多任务动态cron
分组多任务动态cron

1. 普通多任务动态cron 实现

1.1 对应数据库的实体类 TaskEntity


@Data
@AllArgsConstructor
@NoArgsConstructor
public class TaskEntity {
  
  private int taskId;
  
  private String desc;
  
  private String expression;
}

1.2 配置每个任务实现

配置任务接口 TaskService


public interface TaskService {

  void HandlerJob();

  Integer jobId();

}

配置任务接口实现 TaskServiceJob1Impl、TaskServiceJob2Impl …


@Service
public class TaskServiceJob1Impl implements TaskService {
  @Override
  public void HandlerJob() {
    System.out.println("------job1 开始执行---------："+new Date());

    System.out.println(new SimpleDateFORMat("yyyy-MM-dd HH:mm:ss").format(new Date()) + "  " + Thread.currentThread().getName() + "  任务一启动");
    try {
      Thread.sleep(10000);//任务耗时10秒
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + "  " + Thread.currentThread().getName() + "  结束");

  }

  @Override
  public Integer jobId() {
    return 1;
  }
}

1.3 配置任务解析器 TaskSolverChooser

注：
这里引入策略模式
为啥要配置任务解析器选择器:
因为我们实现多个任务时，一个任务对应一个 CronTask，需要在 MyScheduledTask 里面去实现我们每一个方法。
譬如，我们有100个任务就要自定义100个任务实现方法，代码会很臃肿，明显不符合，【开闭原则】，于是这里采用策略模式，解耦我们多个任务业务实现逻辑。


@Slf4j
@Component
public class TaskSolverChooser implements ApplicationContextAware {

  private ApplicationContext applicationContext;

  private Map<Integer, TaskService> chooseMap = new HashMap<>(16);

  
  @Override
  public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    this.applicationContext = applicationContext;
  }

  @PostConstruct
  private void registerToTaskSolver(){
    Map<String, TaskService> taskServiceMap = applicationContext.getBeansOfType(TaskService.class);
    for (TaskService value : taskServiceMap.values()) {
      chooseMap.put(value.jobId(), value);
      log.info("task {} 处理器: {} 注册成功",new Object[]{value.jobId(),value});
    }
  }

  
  public TaskService getTask(Integer jobId){
    return chooseMap.get(jobId);
  }
}

1.4 配置MyScheduledTask （动态cron核心配置）

说明：
1、配置多线程执行任务
2、配置刷新 task
3、配置停止 task
4、配置执行task 业务逻辑


@Component
public class MyScheduledTask implements SchedulingConfigurer {

  private volatile ScheduledTaskRegistrar registrar;

  private final ConcurrentHashMap<Integer, ScheduledFuture<?>> scheduledFutures = new ConcurrentHashMap<>();
  private final ConcurrentHashMap<Integer, CronTask> cronTasks = new ConcurrentHashMap<>();

  @Autowired
  private TaskSolverChooser taskSolverChooser;

  @Override
  public void configureTasks(ScheduledTaskRegistrar registrar) {

    //设置20个线程,默认单线程,如果不设置的话，不能同时并发执行任务
    registrar.setScheduler(Executors.newScheduledThreadPool(10));
    this.registrar = registrar;
  }

  
  public void refresh(List<TaskEntity> tasks){
    //取消已经删除的策略任务
    Set<Integer> sids = scheduledFutures.keySet();
    for (Integer sid : sids) {
      if(!exists(tasks, sid)){
        scheduledFutures.get(sid).cancel(false);
      }
    }
    for (TaskEntity TaskEntity : tasks) {
      String expression = TaskEntity.getExpression();
      //计划任务表达式为空则跳过
      if(!StringUtils.hasLength(expression)){
        continue;
      }
      //计划任务已存在并且表达式未发生变化则跳过
      if (scheduledFutures.containsKey(TaskEntity.getTaskId())
          && cronTasks.get(TaskEntity.getTaskId()).getExpression().equals(expression)) {
        continue;
      }
      //如果策略执行时间发生了变化，则取消当前策略的任务
      if(scheduledFutures.containsKey(TaskEntity.getTaskId())){
        scheduledFutures.get(TaskEntity.getTaskId()).cancel(false);
        scheduledFutures.remove(TaskEntity.getTaskId());
        cronTasks.remove(TaskEntity.getTaskId());
      }
      //业务逻辑处理
      CronTask task = cronTask(TaskEntity, expression);


      //执行业务
      ScheduledFuture<?> future = registrar.getScheduler().schedule(task.getRunnable(), task.getTrigger());
      cronTasks.put(TaskEntity.getTaskId(), task);
      scheduledFutures.put(TaskEntity.getTaskId(), future);
    }
  }

  
  public void stop(List<TaskEntity> tasks){
    tasks.forEach(item->{
      if (scheduledFutures.containsKey(item.getTaskId())) {
        // mayInterruptIfRunning设成false话，不允许在线程运行时中断，设成true的话就允许。
        scheduledFutures.get(item.getTaskId()).cancel(false);
        scheduledFutures.remove(item.getTaskId());
      }
    });
  }

  
  public CronTask cronTask(TaskEntity TaskEntity, String expression) {
    return new CronTask(() -> {
          //每个计划任务实际需要执行的具体业务逻辑
          //采用策略，模式 ，执行我们的job
          taskSolverChooser.getTask(TaskEntity.getTaskId()).HandlerJob();
        }, expression);
  }

  private boolean exists(List<TaskEntity> tasks, Integer tid){
    for(TaskEntity TaskEntity:tasks){
      if(TaskEntity.getTaskId() == tid){
        return true;
      }
    }
    return false;
  }

  @PreDestroy
  public void destroy() {
    this.registrar.destroy();
  }

}

1.5 配置程序启动时首次去加我们设置的task


@Component
public class StartInitTask implements CommandLineRunner {

  @Autowired
  private MyScheduledTask myScheduledTask;

  @Override
  public void run(String... args) throws Exception {
    List<TaskEntity> list = Arrays.asList(
        new TaskEntity(1, "测试1", "0/1 * * * * ?"),
        new TaskEntity(2, "测试2", "0/1 * * * * ?")
    );
    myScheduledTask.refresh(list);
  }
}

1.6 配置web接口去触发，增删启停


@RestController
public class StartController {

  @Autowired
  private MyScheduledTask scheduledTask;

  @PostMapping(value = "/startOrChangeCron")
  public String changeCron(@RequestBody List<TaskEntity> list){
    if (CollectionUtils.isEmpty(list)) {
      // 这里模拟存在数据库的数据
      list = Arrays.asList(
          new TaskEntity(1, "测试1","0/1 * * * * ?") ,
          new TaskEntity(2, "测试2","0/1 * * * * ?")
      );
    }
    scheduledTask.refresh(list);
    return "task任务:" + list.toString() + "已经开始运行";
  }

  @PostMapping(value = "/stopCron")
  public String stopCron(@RequestBody List<TaskEntity> list){
    if (CollectionUtils.isEmpty(list)) {
      // 这里模拟将要停止的cron可通过前端传来
      list = Arrays.asList(
          new TaskEntity(1, "测试1","0/1 * * * * ?") ,
          new TaskEntity(2, "测试2","0/1 * * * * ?")
      );
    }
    scheduledTask.stop(list);
    List<Integer> collect = list.stream().map(TaskEntity::getTaskId).collect(Collectors.toList());
    return "task任务:" + collect.toString() + "已经停止启动";
  }

}

2. 分组多任务动态cron 实现

实现原理：
基于反射实现，根据方法全类名，去动态执行方法。多任务分组配置，根据任务类型进行分组。
eg：
定时任务人员的相关操作，有检测人员离职状态，人员业绩达标，人员考勤…等，
作用：
对人员定时任务做一个分类，在同一个类里面去实现不同的task，
比较
《1. 普通多任务动态cron 实现》，是一个类可以实现一个task
《2. 分组多任务动态cron 实现》，是一个类可以实现多个task
详细可参考：分组多任务动态cron