如何使用ThreadPoolTaskScheduler动态调度Spring Batch作业

midemarc 发表于 Dev

Midemarc

我有一个Spring Batch应用程序，我想在其中计划作业调用。

调度间隔在构建时是未知的，因此我不能仅使用@Scheduled注释作业。这导致我使用ThreadPoolTaskScheduler。

问题是方法进度表将Runnable作为参数。这样可以安排工作吗？我可以通过以下服务直接致电该工作，但无法安排工作。

这是我遇到问题的背景，我试图使其变得简单：

@Service
public class ScheduledProcessor{
    private final ThreadPoolTaskScheduler  threadPoolTaskScheduler;
    private Application application;   
    @Autowired
    public ScheduledProcessor(ThreadPoolTaskScheduler threadPoolTaskScheduler, Application application){
        this.threadPoolTaskScheduler = threadPoolTaskScheduler;
        this.application = application;
        scheduledTasks = new ArrayList();


        Trigger trigger = new CronTrigger("0/6 * * * * *");
//Here I am trying to schedule my job.
//The following line is wrong because a Job can't be cast to a Runnable but I wanted to show the intended behaviour.            
        threadPoolTaskScheduler.schedule((Runnable) application.importUserjob, trigger);
        System.out.println("Job launch !");
    }

这是JobBuilderFactory：

@Bean
public Job importUserJob(JobBuilderFactory jobs, Step s1, Step s2) {
    return jobs.get("importUserJob")
            .incrementer(new RunIdIncrementer())
            .flow(s1)
            .end()
            .build();
}

我了解（嗯，我什至对此不确定）我不能直接将Job转换为Runnable，但是可以以任何方式转换它吗？或者，您可以给我一些有关如何动态调度Spring Batch作业的建议吗？

如果发生了某些变化，我还需要能够重新启动/跳过我的步骤，就像我目前使用threadPoolTaskScheduler一样。

预先感谢您提供的任何帮助或提示。

Midemarc

我终于知道该怎么做了！

我创建了一个实现Runnable的类（为方便起见，扩展了Thread，从而避免了实现所有Runnable类的需要）。

@Component
public class MyRunnableJob extends Thread implements Runnable{

    private Job job;
    private JobParameters jobParameters;
    private final JobOperator jobOperator;

    @Autowired
    public MyRunnableJob(JobOperator jobOperator) {
        this.jobOperator = jobOperator;
    }

    public void setJob(Job job){
        this.job=job;
    }

    @Override
    public void run(){
        try {
            String dateParam = new Date().toString();
            this.jobParameters = new JobParametersBuilder().addString("date", dateParam).toJobParameters();
            System.out.println("jobName : "+job.getName()+" at "+dateParam);

            jobOperator.start(job.getName(), jobParameters.toString());
        } catch (NoSuchJobException | JobInstanceAlreadyExistsException | JobParametersInvalidException ex) {
            Logger.getLogger(MyRunnableJob.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
}

在我的ScheduledProcessor类中，将Job设置为myRunnable类，然后将其作为schedule方法的参数传递。

public class SchedulingProcessor {

    //Autowired fields :
    private final JobLauncher jobLauncher;
    private final Job importUserJob;
    private final ThreadPoolTaskScheduler  threadPoolTaskScheduler;
    private final MyRunnableJob myRunnableJob;


    //Other fields :
    private List<ScheduledFuture> scheduledTasks;


    @Autowired
    public SchedulingProcessor(JobLauncher jobLauncher, Job importUserJob, ThreadPoolTaskScheduler  threadPoolTaskScheduler, MyRunnableJob myRunnableJob) throws Exception {
        this.jobLauncher=jobLauncher;
        this.importUserJob=importUserJob;
        this.threadPoolTaskScheduler=threadPoolTaskScheduler;
        this.myRunnableJob=myRunnableJob;
        Trigger trigger = new CronTrigger("0/6 * * * * *");
        myRunnableJob.setJob(this.importUserJob);
        scheduledTasks = new ArrayList();
        scheduledTasks.add(this.threadPoolTaskScheduler.schedule((Runnable) myRunnableJob, trigger));
    }        
}