SpringBoot单体任务调度

一、TaskScheduler

• 我们对应cron表达式要用到的方法

public interface TaskScheduler {
/**调度器通过给定的 Runnable，
每当触发器 到指示下一个执行时间时调用它。
*/
@Nullable
	ScheduledFuture<?> schedule(Runnable task, Trigger trigger);
	...
	}

• Trigger用于计算任务的下次执行时间

public interface Trigger {
// 有触发器定义下次执行时间，或不再触发
    @Nullable
    Date nextExecutionTime(TriggerContext triggerContext);

}

• 其有两个实现（CronTrigger、PeriodicTrigger）
• CronTrigger通过Cron表达式来生成调度计划。
• PeriodicTrigger用于间隔时间来定期执行

二、调度器管理者

@Component
public class CronTaskManager implements DisposableBean {
    private final Map<String, ScheduleInfo> scheduledTaskMap = new ConcurrentHashMap<>(8);

    @Autowired
    private TaskScheduler taskScheduler;

    public TaskScheduler getScheduler() {
        return this.taskScheduler;
    }
// 增加任务
    public void addCronTask(String jobId, Runnable runnable, String cronExpression) {
        CronTask cronTask = new CronTask(runnable, cronExpression);
        if (this.scheduledTaskMap.containsKey(jobId)) {
            removeCronTask(jobId);
        }
        ScheduleInfo schedulerInfo = new ScheduleInfo(jobId, runnable, scheduleCronTask(cronTask));
        scheduledTaskMap.put(jobId, schedulerInfo);
    }

// 删除任务
    public void removeCronTask(String jobId) {
        ScheduleInfo scheduleInfo = this.scheduledTaskMap.remove(jobId);
        if (Objects.nonNull(scheduleInfo)) {
            ScheduledTaskFuture scheduledTask = scheduleInfo.getScheduledTaskFuture();
            scheduledTask.cancel();
        }
    }
// 调度任务定义
    public ScheduledTaskFuture scheduleCronTask(CronTask cronTask) {
        ScheduledTaskFuture scheduledTask = new ScheduledTaskFuture();
        scheduledTask.future = this.taskScheduler.schedule(cronTask.getRunnable(), cronTask.getTrigger());
        return scheduledTask;
    }

// 最后spring结束时，取消定时任务并清空任务列
    @Override
    public void destroy() {
        for (ScheduleInfo scheduleInfo : this.scheduledTaskMap.values()) {
            ScheduledTaskFuture task = scheduleInfo.getScheduledTaskFuture();
            task.cancel();
        }
        this.scheduledTaskMap.clear();
    }
}

• ConcurrentHashMap可以看看

• 通过ScheduledFuture实现任务取消

public final class ScheduledTaskFuture {
// volatile 保证了不同线程对共享变量操作的可见性，也就是说一个线程修改了 volatile 修饰的变量，当修改后的变量写回主内存时，其他线程能立即看到最新值。
    volatile ScheduledFuture<?> future;

    /**
     * 取消定时任务
     */
    public void cancel() {
        ScheduledFuture<?> future = this.future;
        if (future != null) {
            future.cancel(true);
        }
    }
}

三、任务调度的线程池配置

@Configuration
public class ScheduleThreadPoolConfig {

    @Bean
    public TaskScheduler taskScheduler() {
        ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
        //核心线程数
        threadPoolTaskScheduler.setPoolSize(8);
        //设置删除取消策略
        threadPoolTaskScheduler.setRemoveOnCancelPolicy(true);
        //前缀
        threadPoolTaskScheduler.setThreadNamePrefix("myPool-");
        threadPoolTaskScheduler.initialize();
        return threadPoolTaskScheduler;
    }
}

四、任务加载

• ApplicationRunner容器启动完成时立马加载任务

@Component
public class TaskLoadingRunner implements ApplicationRunner {

    @Autowired
    private Service service;

    @Autowired
    private CronTaskManager cronTaskManager;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        //获取所有任务数据
        List<Entity> list= service.selectAll();
        //将核检任务结合bean name以及cron表达式创建对应调度任务
        if (!CollectionUtils.isEmpty(list)) {
            list.forEach(x -> {
                ScheduleRunnable scheduleRunnable = new ScheduleRunnable(x.getId(), x.getBeanName());
                // 验证cron合规
                if (CronUtil.validCronExpression(x.getCronExpression())) {
                    try {
                        cronTaskManager.addCronTask(x.getId(), scheduleRunnable, x.getCronExpression());
                    } catch (Exception e) {
                        System.out.println("加载任务异常：{}", e.getMessage());
                    }
                }
            });
        }
    }

• ScheduleRunnable实现run方法的任务执行类