Java 如何触发预定的 Spring Batch 作业？

How to trigger a scheduled Spring Batch Job?

提问by akcasoy

I want to be able to start my job with a REST controller, then when the job is started, it should run on a scheduled basis, until i stop it again with REST.

我希望能够使用 REST 控制器开始我的工作，然后当工作开始时，它应该按计划运行，直到我再次使用 REST 停止它。

So this is my Controller:

所以这是我的控制器：

@RestController
public class LauncherController {

    @Autowired
    JobLauncher jobLauncher;

    @Autowired
    Job job;

    @RequestMapping("/launch")
    public String launch() throws Exception {
             ...
            jobLauncher.run(job, jobParameters);
    }

This is some part of the Batch conf:

这是 Batch conf 的一部分：

@Configuration
@EnableBatchProcessing
@EnableScheduling
public class BatchConfiguration {

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    @Scheduled(cron = "0/5 * * * * ?")
    @Bean
    public Job job() {
        return jobBuilderFactory.get("job")
                .incrementer(new RunIdIncrementer())
                .flow(step1())
                .end()
                .build();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                .<Person, Person> chunk(10)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .build();
    }

I have also set the property spring.batch.job.enabled=false, since i do not want the jobs run as soon as the Spring Boot App starts.

我还设置了属性 spring.batch.job.enabled=false，因为我不希望在 Spring Boot 应用程序启动后立即运行作业。

Now i can call my Rest api lauch, and the job runs, but only once. Scheduler does not work. And I could not figure it our where exactly i should define my @Scheduled Annotation..

现在我可以调用我的 Rest api lauch，并且作业运行，但只有一次。调度程序不起作用。我无法弄清楚我应该在哪里定义我的@Scheduled Annotation..

采纳答案by Maciej Walkowiak

I would approach it in a way, that scheduled job runs always, but it does something only when the flag is set to true:

我会以某种方式接近它，预定的作业总是运行，但只有当标志设置为 true 时它才会执行某些操作：

@Component
class ScheduledJob {

    private final AtomicBoolean enabled = new AtomicBoolean(false);

    @Scheduled(fixedRate = 1000)
    void execute() {
        if (enabled.get()) {
            // run spring batch here.
        }
    }

    void toggle() {
        enabled.set(!enabled.get());
    }

}

and a controller:

和控制器：

@RestController
class HelloController {

    private final ScheduledJob scheduledJob;

    // constructor

    @GetMapping("/launch")
    void toggle() {
        scheduledJob.toggle();
    }

}

回答by user7294900

@Scheduledis defined on a method and not on a Bean. So create a new Class which will be a Bean

@Scheduled是在方法上定义的，而不是在 Bean 上定义的。所以创建一个新的类，它将是一个 Bean

public class BatchConfiguration {
...
@Bean
public Job job() {
    return new Job();
}

new Class:

新班级：

public class Job {

@Scheduled(cron = "0/5 * * * * ?")
public Job job() {
    return jobBuilderFactory.get("job")
            .incrementer(new RunIdIncrementer())
            .flow(step1())
            .end()
            .build();
}

回答by Patrick

In this solution you will be able to schedule and unschedule pre defined jobs using http requests. In this example we will create a daily, weekly and an oneTime Job. The application is using Quartz.

在此解决方案中，您将能够使用 http 请求安排和取消安排预定义的作业。在此示例中，我们将创建每日、每周和一次性作业。该应用程序正在使用Quartz.

<!--Quartz Scheduler -->
<dependency>
    <groupId>org.quartz-scheduler</groupId>
    <artifactId>quartz</artifactId>
    <version>2.2.3</version>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-tx</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context-support</artifactId>
</dependency>

First we have to create an AutowiringSpringBeanJobFactoryclass extends SpringBeanJobFactory.

首先，我们必须创建一个AutowiringSpringBeanJobFactory类 extends SpringBeanJobFactory。

• Subclass of {@link AdaptableJobFactory} that also supports Spring-style * dependency injection on bean properties. This is essentially the direct * equivalent of Spring's {@link QuartzJobBean} in the shape of a Quartz * {@link org.quartz.spi.JobFactory}. * *

  Applies scheduler context, job data map and trigger data map entries * as bean property values. If no matching bean property is found, the entry * is by default simply ignored. This is analogous to QuartzJobBean's behavior.

• {@link AdaptableJobFactory} 的子类，它也支持对 bean 属性的 Spring 式 * 依赖注入。这本质上是 Quartz * {@link org.quartz.spi.JobFactory} 形状的 Spring {@link QuartzJobBean} 的直接 * 等价物。* *

  应用调度程序上下文、作业数据映射和触发器数据映射条目 * 作为 bean 属性值。如果未找到匹配的 bean 属性，则默认情况下会忽略条目 *。这类似于 QuartzJobBean 的行为。

public final class AutowiringSpringBeanJobFactory extends SpringBeanJobFactory implements ApplicationContextAware {

    private transient AutowireCapableBeanFactory beanFactory;

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

    @Override
    protected Object createJobInstance(final TriggerFiredBundle bundle) throws Exception {
        final Object job = super.createJobInstance(bundle);
        beanFactory.autowireBean(job);
        return job;
    }
}

The second part is to configure the quartz configuration. In this config we need to create a

第二部分是配置quartz配置。在这个配置中，我们需要创建一个

• SchedulerFactoryBeanwhere we set global config and the application context,

• JobDetailFactoryBeanwhere we set our job, the jobGroup and the class,

• CronTriggerFactoryBeanwhere we set the cron expression.

• SchedulerFactoryBean我们在这里设置全局配置和应用程序上下文，

• JobDetailFactoryBean我们在哪里设置我们的工作，jobGroup 和类，

• CronTriggerFactoryBean我们在这里设置 cron 表达式。

QuartzConfig.class

QuartzConfig.class

@Configuration
public class QuartzConfig {

    @Autowired
    ApplicationContext context;

    @Bean
    public SchedulerFactoryBean quartzScheduler(){
        SchedulerFactoryBean quartzScheduler = new SchedulerFactoryBean();
        quartzScheduler.setOverwriteExistingJobs(true);
        quartzScheduler.setSchedulerName("job-scheduler");
        AutowiringSpringBeanJobFactory jobFactory = new AutowiringSpringBeanJobFactory();
        jobFactory.setApplicationContext(context);
        quartzScheduler.setJobFactory(jobFactory);
        return quartzScheduler;
    }

    @Bean
    @Scope(value = "prototype")
    public JobDetailFactoryBean getJobBean(String jobName, String jobGroup, Class<?> clazz){
        JobDetailFactoryBean bean = new JobDetailFactoryBean();
        bean.setJobClass(clazz);
        bean.setGroup(jobGroup);
        bean.setName(jobName);
        return bean;
    }

    @Bean
    @Scope(value = "prototype")
    public CronTriggerFactoryBean getCronTriggerBean(String cronExpression, String triggerGroup){
        CronTriggerFactoryBean bean = new CronTriggerFactoryBean();
        bean.setCronExpression(cronExpression);
        bean.setGroup(triggerGroup);
        return bean;
    }
}

So, after the config is done we are now able to create our jobs where the business logic will be placed. For that we have to create a class which implements Job.

因此，在配置完成后，我们现在可以创建将放置业务逻辑的作业。为此，我们必须创建一个实现Job.

@Component
public class DailyJob implements Job{

    @Override
    public void execute(JobExecutionContext context) throws JobExecutionException {
        System.out.println("Daily Job runs!");
    }
}

The DailyJobclass is now ready to get scheduled. We want to schedule this job from outside via a http request. In this example we have a controller where we can send the jobname and the cron expression to schedule the dailyJob.

该DailyJob班是现在准备好计划。我们想通过 http 请求从外部安排这项工作。在这个例子中，我们有一个控制器，我们可以在其中发送作业名和 cron 表达式来安排dailyJob.

@Controller
public class JobController {

    @Autowired
    private Scheduler scheduler;
    @Autowired
    private ApplicationContext context;;

    @ResponseBody
    @RequestMapping(value = "/job/create/daily", method = RequestMethod.POST)
    public ResponseEntity<JobModel> dailyJob(@RequestBody JobModel jobModel) throws SchedulerException {
        JobDetail jobDetail = context.getBean(
                JobDetail.class, jobModel.getName(), "MyDailyJob", DailyJob.class);
        Trigger cronTrigger = context.getBean(
                Trigger.class, jobModel.getCronExpression(), "MyDailyJob");

        scheduler.scheduleJob(jobDetail, cronTrigger);

        return new ResponseEntity<JobModel>(jobModel, HttpStatus.CREATED);
    }
}

What we see here is that we will send a post request with a JobModelas @RequestBody. JobModelis a simple Pojo with two attributes nameand cronExpressionboth Strings.

我们在这里看到的是，我们将发送一个带有JobModelas的 post 请求@RequestBody。JobModel是一个简单的 Pojo，具有两个属性name和cronExpression两个字符串。

In this method we have to create the bean instances which we have configured previously in our config class. First create JobDetailwith Quartz JobDetail.class, the name of your job, the name of the group and the Class which should be scheduled (in this case DailyJob.class). After that we have to create the Trigger with Quartz Trigger.class, the cronExpression and the group name.

在这个方法中，我们必须创建我们之前在配置类中配置的 bean 实例。首先JobDetail使用 Quartz创建JobDetail.class您的工作名称、组名称和应该安排的类（在本例中为DailyJob.class）。之后，我们必须使用 Quartz Trigger.class、cronExpression 和组名创建触发器。

After both beans are created we need to schedule the job now. So we have autowired Quartz Schedulerto schedule the job. After that the job is enabled and ready to do its job.

创建两个 bean 后，我们现在需要安排作业。所以我们已经自动装配了 QuartzScheduler来安排工作。之后，该作业已启用并准备好完成其工作。

So let's test the stuff. Start the application and send a post request to /job/create/daily:

所以让我们测试一下这些东西。启动应用程序并将发布请求发送到/job/create/daily：

{"name":"Job 1", "cronExpression":"0 * * * * ?"}

Here we say that the job should run every minute (just to see that everything works). In your console you should see every minute Daily Job runs!.

这里我们说作业应该每分钟运行一次（只是为了看看一切正常）。在您的控制台中，您应该看到每一分钟Daily Job runs!。

And here are some additional things you can do. For example get a list of the scheduled jobs:

这里还有一些您可以做的其他事情。例如获取预定作业的列表：

 @ResponseBody
 @RequestMapping("job/list")
 public List<String> jobList() throws SchedulerException {
     return scheduler.getJobGroupNames();
 }

To delete a job you can create endpoints, too. For example:

要删除作业，您也可以创建端点。例如：

@ResponseBody
@RequestMapping(value = "job/delete/daily", method = RequestMethod.POST)
public ResponseEntity<Boolean> deleteJob(@RequestBody JobModel jobModel) throws SchedulerException {
    JobKey jobKey = new JobKey(jobModel.getName(), "MyDailyJob");
    return new ResponseEntity<Boolean>(scheduler.deleteJob(jobKey), HttpStatus.OK);
}

You are free to create many different endpoints to get informations about currently running jobs, how often jobs were running, reschedule jobs and so on. Important is just, that your jobname and the jobgroup( in our case "MyDailyJob") are reusable. Those information are needed to create the jobKey.

您可以自由创建许多不同的端点来获取有关当前正在运行的作业、作业运行的频率、重新安排作业等的信息。重要的是，您的工作名称和工作组（在我们的例子中"MyDailyJob"）是可重用的。创建 jobKey 需要这些信息。

P.S.: Just to show the other mappings for the other jobs:

PS：只是为了显示其他作业的其他映射：

@ResponseBody
@RequestMapping(value = "/job/create/weekly", method = RequestMethod.POST)
public ResponseEntity<JobModel> weeklyJob(@RequestBody JobModel jobModel) throws SchedulerException {
    JobDetail jobDetail = context.getBean(JobDetail.class, jobModel.getName(), JobGroup.WEEKLY_GROUP.name(),
            WeeklyJob.class);
    Trigger cronTrigger = context.getBean(Trigger.class, jobModel.getCronExpression(),
            JobGroup.WEEKLY_GROUP.name());

    scheduler.scheduleJob(jobDetail, cronTrigger);

    return new ResponseEntity<JobModel>(jobModel, HttpStatus.CREATED);

}

@ResponseBody
@RequestMapping(value = "/job/create/oneTime", method = RequestMethod.POST)
public ResponseEntity<JobModel> oneTimeJob(@RequestBody JobModel jobModel) throws SchedulerException {
    JobDetail jobDetail = context.getBean(JobDetail.class, jobModel.getName(), JobGroup.ONE_TIME_GROUP.name(),
            OneTimeJob.class);
    Trigger cronTrigger = context.getBean(Trigger.class, jobModel.getCronExpression(),
            JobGroup.ONE_TIME_GROUP.name());

    scheduler.scheduleJob(jobDetail, cronTrigger);

    return new ResponseEntity<JobModel>(jobModel, HttpStatus.CREATED);
}

The full application is on github

完整的应用程序在github上

回答by Ilya Dyoshin

In first you are defining the job:

首先，您要定义工作：

@Bean
@Qualifier("fancyScheduledJob")
public Job job() {
    return jobBuilderFactory.get("job")
            .incrementer(new RunIdIncrementer())
            .flow(step1())
            .end()
            .build();
}

In second you are initiating the execution of this job:

其次，您正在启动此作业的执行：

@Autowired
@Qualifier(value = "fancyScheduledJob")
private Job job;

@Autowired
private JobLauncher jobLauncher;

@Scheduled(cron = "0/5 * * * * ?")
public void launch() throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException, JobInstanceAlreadyExistsException, NoSuchJobException {

    jobLauncher.run(job, JobParametersBuilder()
            .addLong("launchTime", System.currentTimeMillis())
            .toJobParameters())
}

Also note that the "launchTime" paramter is introduced: by default spring batch is preventing launching the job with same parameter values.

另请注意，引入了“launchTime”参数：默认情况下，spring 批处理阻止启动具有相同参数值的作业。

While your schedule is quite tight - every 5 seconds you should be aware of concurrency. Or if you want to be assured that at each and every moment only 1 instance of the job is executed you can configure custom single threaded job launcher:

虽然您的日程安排很紧 - 每 5 秒您应该意识到并发性。或者，如果您想确保每时每刻只执行 1 个作业实例，您可以配置自定义单线程作业启动器：

@Bean(name = "fancyJobExecutorPool")
public TaskExecutor singleThreadedJobExecutorPool() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(1);
    executor.setMaxPoolSize(1);
    executor.setQueueCapacity(100500);
    executor.setThreadNamePrefix("fancy-job-batch-");
    return executor;
}

@Bean(name = "fancyJobLauncher")
public JobLauncher singleThreadedJobLauncher(JobRepository jobRepository)
{
    SimpleJobLauncher sjl = new SimpleJobLauncher();
    sjl.setJobRepository(jobRepository);
    sjl.setTaskExecutor(singleThreadedJobExecutorPool());
    return sjl;
}

And use this single threaded job launcher during launch time.

并在启动时使用这个单线程作业启动器。

@Autowired
@Qualifier("fancyJobLauncher")
private JobLauncher jobLauncher;

With this your job instances will be executed one by one (but this doesn't limits parallel execution of steps inside of your job).

有了这个，您的作业实例将被一个一个地执行（但这并不限制您的作业内部步骤的并行执行）。