Java Spring Boot 计划作业

Question

我有一个包含一些作业的应用程序，我想将这些作业保留在数据库中以便更好地维护。

该应用程序的结构如下:

数据库:

CREATE TABLE jobs_config (
    id int8 NOT NULL GENERATED BY DEFAULT AS IDENTITY,
    job_name varchar NOT NULL,
    scheduled_value varchar NOT NULL,
    CONSTRAINT jobs_config_pk PRIMARY KEY (id),
    CONSTRAINT jobs_config_name_un UNIQUE (job_name)
);
insert into jobs_config(job_name, scheduled_value) values('DeleteExpiredTokenJob', '0 0 10 * * MON');

JAVA Spring 启动:

@Entity
@Table(name = "jobs_config")
public class JobsConfig {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(name = "job_name")
    private String jobName;

    private String scheduledValue;

    public JobsConfig() {
    }

    public JobsConfig(String jobName, String scheduledValue) {
        this.jobName = jobName;
        this.scheduledValue = scheduledValue;
    }

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getJobName() {
        return jobName;
    }

    public void setJobName(String jobName) {
        this.jobName = jobName;
    }

    public String getScheduledValue() {
        return scheduledValue;
    }

    public void setScheduledValue(String scheduledValue) {
        this.scheduledValue = scheduledValue;
    }
}

@Service
public class JobSchedulerService implements SchedulingConfigurer {

    private static Logger logger = LoggerFactory.getLogger(JobSchedulerService.class);

    @Autowired
    JobsConfigRepository jobsConfigRepository;

    @Autowired
    DeleteExpiredTokenJob deleteExpiredTokenJob;

    @Autowired
    TestJob testJob;

    @Bean
    public TaskScheduler poolScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setThreadNamePrefix("ThreadPoolTaskScheduler");
        scheduler.setPoolSize(1);
        scheduler.initialize();
        return scheduler;
    }

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        taskRegistrar.setTaskScheduler(poolScheduler());
        listJobList();
//        refreshJobList( poolScheduler());
    }

    public void listJobList() {
        List<JobsConfig>  jobsList = jobsConfigRepository.findAll();
        for (JobsConfig jobName : jobsList) {
            switch (jobName.getJobName()) {
                case "DeleteExpiredTokenJob":
                    scheduleJob(poolScheduler(), deleteExpiredTokenJob, jobName.getJobName());
                    break;
                case "TestJob":
                    scheduleJob(poolScheduler(), testJob, jobName.getJobName());
                    break;
                default:
                    logger.info(String.format("JOB NOT FOUND [%s]", jobName.getJobName()));
            }
        }
    }

    public void scheduleJob(TaskScheduler scheduler, JobInterface jobInterface, String jobName){
        scheduler.schedule(new Runnable(){
            @Override
            public void run() {
                jobInterface.jobCode();
            }
        }, new Trigger(){
            @Override
            public Date nextExecutionTime(TriggerContext triggerContext) {
                Optional <JobsConfig> job = jobsConfigRepository.findByJobName(jobName);
                String cronExp = job.get().getScheduledValue();
                return new CronTrigger(cronExp).nextExecutionTime(triggerContext);
            }
        });
    }
    //THIS IS WHAT I'VE TRIED !
    /*private void refreshJobList(TaskScheduler scheduler){
        scheduler.schedule(new Runnable(){
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName()+" The Task2 executed at "+ new Date());
                listJobList();
            }
        }, new Trigger(){
            @Override
            public Date nextExecutionTime(TriggerContext triggerContext) {
                String cronExp="0/10 * * * * ?";//Can be pulled from a db . This will run every minute
                return new CronTrigger(cronExp).nextExecutionTime(triggerContext);
            }
        });
    }*/

}

这个想法是，每次我在 jobs_config 表中添加一条记录时，我都需要将其反射(reflect)在 java 中。

因此，如果我在表中添加新作业，我希望无需重新启动应用程序即可使用(当然，DeleteExpiredTokenJob 的 java 代码存在)。

insert into jobs_config(job_name, scheduled_value) values('DeleteExpiredTokenJob', '0/5 * * * * ?');

基本上我需要刷新列表List<JobsConfig> jobsList = jobsConfigRepository.findAll();来自listJobList()方法。

我怎样才能实现这个目标？

Answer 1

我发现您尝试创建一个可以刷新作业的作业，但没有成功。

我看不到在数据库中添加新作业后使其可用的简单方法，假设它有一个新的@Autowired作业要注入(inject)，但我想我知道如何跟踪数据库中的更新作业，即scheduled_value列。

1. 将旧的 JobsConfig 值保留在此处。

2. 检查其中是否有任何更改。

   2.a。如果发生更改，则取消下一个作业执行，并安排更新的作业。

3. 睡几分钟以进行下一次检查。 (就像您在注释代码中所做的那样)

    Map<String, JobsConfig> oldJobsConfigs = new HashMap<>();
    Map<String, ScheduledFuture<?>> activeJobs = new HashMap<>()

    public void listJobList() {
        List<JobsConfig> jobsList = jobsConfigRepository.findAll();
        for (JobsConfig jobName : jobsList) {
            // If this job was there before and has not changed, do nothing.
            if (oldJobsConfigs.containsKey(jobName.getJobName()) && oldJobsConfigs.get(jobName.getJobName()).getScheduledValue().equals(jobName.getScheduledValue())) 
                break;

            // Cancel previous execution, if any.
            if (activeJobs.containsKey(jobName.getJobName()) {
                ScheduledFuture<?> job = activeJobs.get(jobName.getJobName());
                job.cancel(false);
                try {
                    job.get(); // Warning! If the job is running, blocks current thread until the job finishes. If has an endless loop, it will block current thread forever.
                } catch (CancellationException e) {
                    // Do nothing, this is good, we did not spent time waiting for the job to finish.
                } catch (InterruptedException | ExecutionException e) {
                    // Log it?
                }
            }

            ScheduledFuture<?> newJob = null;
            switch (jobName.getJobName()) {
                case "DeleteExpiredTokenJob":
                    newJob = scheduleJob(poolScheduler(), deleteExpiredTokenJob, jobName.getJobName());
                    break;
                case "TestJob":
                    newJob = scheduleJob(poolScheduler(), testJob, jobName.getJobName());
                    break;
                default:
                    logger.info(String.format("JOB NOT FOUND [%s]", jobName.getJobName()));
            }

            if (newJob != null)
                activeJobs.put(jobName.getJobName(), newJob);
        }
    }

并更改 scheduleJob 签名，以便它将利用 scheduler 返回的内容。

    public ScheduledFuture<?> scheduleJob(TaskScheduler scheduler, JobInterface jobInterface, String jobName) {
        return scheduler.schedule(new Runnable() {
// ... unchanged

希望有帮助。 :D

UPD:如果有一天您的作业数量远远超过十个，您可以稍微调整一下 JobInterface ，以便它将返回其作业名称:

    public interface JobInterface {
        // ... old methods
        String getJobName(); // Consider switching to enums? 
    }

并让 Spring 将 JobInterface 的所有实现 Autowiring 到 JobSchedulerService:

@Service
public class JobSchedulerService implements SchedulingConfigurer {
    @Autowired
    JobsConfigRepository jobsConfigRepository;

    @Autowired
    List<JobInterface> allJobs;

然后，您将编写的不是包含大量条目的 switch

    ScheduledFuture<?> newJob = null;
    for(JobInterface job : allJobs)
        if (job.getJobName().equals(jobName.getJobName()))
            newJob = scheduleJob(/*arguments*/);

    if (newJob == null)
        logger.warn(/*swear loudly :)*/);

也就是说，仅当您想要删除此开关时。如果您打算工作的数量很少，那么保持原样也可以。