Spring Boot - 停止动态启动的任务

Question

用户可以在特定日期后通过 cron 表达式通过用户界面动态启动一些特定任务。任务存储在数据库中，以备需要重新启动时使用。

在研究一些替代方案时，我发现了 ScheduledTaskRegistrar 的解决方案。看起来不错。我的代码的主要部分如下。

我的问题是：如何停止上下文中任何最初/cron 启动的作业？

我查看了与 ScheduledTaskRegistrar 相关的类，但尚未找到任何解决方案。使用 Quartz 是另一种解决方案，但我更喜欢轻量级、简单的解决方案。

@Override
public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
  this.taskRegistrar = taskRegistrar;
  LocalDateTime now = LocalDateTime.now();
  taskDefinitionRepository.findAll().forEach( t -> {
    if( t.getFirstDateTime().isBefore( now)) {
      taskRegistrar.addTriggerTask( exampleTaskBean, triggerContext -> {
        if( triggerContext.lastScheduledExecutionTime() == null) {
          return convertLocalDateTimeToDate( t.getFirstDateTime());
        } else {
          taskRegistrar.addCronTask( exampleTaskBean, t.getThenCronExpression());
          return null;
        }
      });
    } else {
      taskRegistrar.addCronTask( exampleTaskBean, t.getThenCronExpression());
    }
  });
}

有办法停止特定任务吗？任务不会很多。

是否有更好的简单框架或者 Quartz 是正确的选择？

Answer 1

在调查现有的 Quartz 库时，（使用 iso 创建）该包不适合我的情况。它的数据库表比我的整个应用程序还要多。

在大约 15 分钟内，我编写了以下包。它对于使用 Cron 表达式调度任务来说已经足够好了。崩溃后，任务也会自动重新安排。

配置：

@Configuration
@EnableScheduling
public class TaskSchedulingConfig implements SchedulingConfigurer {

  public static final int SCHEDULER_JOB_POOL_SIZE = 5;

  @Override
  public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
    taskRegistrar.setTaskScheduler(taskScheduler());
  }

  @Bean
  public TaskScheduler taskScheduler() {
    ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
    scheduler.setPoolSize(SCHEDULER_JOB_POOL_SIZE);
    scheduler.setThreadNamePrefix("SomeThreadScheduler__");
    scheduler.initialize();
    return scheduler;
  }
}

任务定义：

@Builder
@NoArgsConstructor
@AllArgsConstructor
@Data
@Entity
@Table(name = "LTST_SCHEDULED_TASK")
public class TaskDefinition {
  @Id
  @GeneratedValue(strategy = GenerationType.AUTO)
  @Column(name = "LTST_ID", updatable = false, nullable = false)
  private Long id;

  @Column(name = "LTST_TASK_NAME")
  private String taskName;

  @Column(name = "LTST_CRON_EXPR")
  private String cronExpression;

  @Column(name = "LTST_TASK_DATA")
  private String taskData;
}

样本服务。使用原型 bean 当然会更好，但就我而言，这已经足够好了。您可以轻松地增强这一点。

@Service
public class TaskDefinitionBean1 implements Runnable {
  private final Logger logger = LoggerFactory.getLogger(TaskDefinitionBean1.class);

  @Override
  public void run() {
    logger.info("Running job task-1");
  }
}

调度服务，基于之前的配置：

@Service
public class TaskSchedulingService {
  private static final Logger logger = LoggerFactory.getLogger(TaskSchedulingService.class);

  private final TaskScheduler taskScheduler;
  private final TaskDefinitionBean1 taskBean1;
  private final TaskDefinitionBean2 taskBean2;
  private final TaskDefinitionRepository taskDefinitionRepository;
  private final Map<String, ScheduledFuture<?>> schedulerMap = new HashMap<>();

  public TaskSchedulingService(TaskScheduler taskScheduler, TaskDefinitionBean1 taskBean1,
                               TaskDefinitionBean2 taskBean2, TaskDefinitionRepository taskDefinitionRepository) {
    this.taskScheduler = taskScheduler;
    this.taskBean1 = taskBean1;
    this.taskBean2 = taskBean2;
    this.taskDefinitionRepository = taskDefinitionRepository;
    startScheduling();
  }

  private void startScheduling() {
    taskDefinitionRepository.findAll().forEach(t -> {
      logger.info("Start existing task '{}' with cron expression {}", t.getTaskName(), t.getCronExpression());
      // Simplistic way to start different services ... good enough for prototyping
      if( t.getTaskName().matches( "one")) {
        schedulerMap.put(t.getTaskName(), taskScheduler.schedule(taskBean1, new CronTrigger(t.getCronExpression())));
      } else {
        schedulerMap.put(t.getTaskName(), taskScheduler.schedule(taskBean2, new CronTrigger(t.getCronExpression())));
      }
    });
  }

  public TaskDefinition addCronTask(String taskName, String cronExpression) {
    if (schedulerMap.containsKey(taskName)) {
      return null;
    }
    TaskDefinition taskDefinition = TaskDefinition.builder().taskName(taskName).cronExpression(cronExpression).build();
    logger.info("Start new task '{}' with cron expression {}", taskDefinition.getTaskName(), taskDefinition.getCronExpression());
    // Simplistic way to start different services ... good enough for prototyping
    if( taskDefinition.getTaskName().matches( "one")) {
      schedulerMap.put(taskDefinition.getTaskName(), taskScheduler.schedule(taskBean1, new CronTrigger(taskDefinition.getCronExpression())));
    } else {
      schedulerMap.put(taskDefinition.getTaskName(), taskScheduler.schedule(taskBean2, new CronTrigger(taskDefinition.getCronExpression())));
    }
    taskDefinitionRepository.save(taskDefinition);
    return taskDefinition;
  }

  public void stopCronTask(String taskName) {
    ScheduledFuture<?> scheduleJob = schedulerMap.get(taskName);
    if (scheduleJob == null) {
      return; // unknow job, so don't stop
    }
    scheduleJob.cancel(false);
    taskDefinitionRepository.deleteByTaskName( taskName);
    schedulerMap.remove(taskName);
  }

  public List<TaskDefinition> getScheduledTasks() {
    return StreamSupport.stream(taskDefinitionRepository.findAll().spliterator(), false).toList();
  }

}

将其放在 API 后面，以便您可以自己创建演示：

@RestController
@RequestMapping("/scheduledtasks")
public class ScheduleTaskController {
  private static final Logger logger = LoggerFactory.getLogger(ScheduleTaskController.class);
  private final TaskSchedulingService taskSchedulingService;

  public ScheduleTaskController(TaskSchedulingService taskSchedulingService) {
    this.taskSchedulingService = taskSchedulingService;
  }

  @GetMapping(value = "")
  public ResponseEntity<List<TaskRequest>> getScheduleTasks() {
    return ok().body( taskSchedulingService.getScheduledTasks().stream().map(this::convertTaskDefinitionToTaskRequest).toList());
  }

  @PostMapping(value = "", consumes = "application/json")
  public ResponseEntity<TaskDefinition> startTaskWithName(@RequestBody TaskRequest taskRequest) {
    logger.info("Start task with name {} and cronexpression {}", taskRequest.getTaskName(),
      taskRequest.getCronExpression());
    return status(HttpStatus.OK).body( taskSchedulingService.addCronTask( taskRequest.getTaskName(),
      taskRequest.getCronExpression()));
  }

  @DeleteMapping(value = "/{taskname}")
  public ResponseEntity<TaskRequest> deleteTaskWithName(@PathVariable("taskname") String taskname) {
    logger.info("Delete task with name {}", taskname);
    taskSchedulingService.stopCronTask( taskname);
    return status(HttpStatus.OK).body( new TaskRequest( taskname, ""));
  }

  private TaskRequest convertTaskDefinitionToTaskRequest( TaskDefinition taskDefinition) {
    return new TaskRequest(taskDefinition.getTaskName(), taskDefinition.getCronExpression());
  }

}

最后，一种用于任务的 POJO：

@AllArgsConstructor
@Data
public class TaskRequest {
  private String taskName;
  private String cronExpression;
}