SpringBatch(14):JobOperator启动Job

前言

我们可以使用JobLaucher来启动Job。需要注入JobLaucher和具体的Job实例

我们也可以使用JobOperator来启动Job,而且操作起来更加简单。

第一节 配置JobOperatorConfig

1. 配置JobOperatorConfig, 将Job操作相关的Bean注入给JobOperator ,并注入Spring的上下文对象。

package com.it2.springbootspringbatch01.joboperator;

import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.configuration.JobRegistry;
import org.springframework.batch.core.configuration.support.JobRegistryBeanPostProcessor;
import org.springframework.batch.core.converter.DefaultJobParametersConverter;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.JobOperator;
import org.springframework.batch.core.launch.support.SimpleJobOperator;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@Slf4j
public class MyJobOperatorConfig implements ApplicationContextAware {

    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private JobExplorer jobExplorer;

    @Autowired
    private JobRegistry jobRegistry;

    @Autowired
    private JobRepository jobRepository;

    private ApplicationContext context;

    @Bean
    public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor() throws Exception {
        JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor=new JobRegistryBeanPostProcessor();
        jobRegistryBeanPostProcessor.setJobRegistry(jobRegistry);
        jobRegistryBeanPostProcessor.setBeanFactory(context.getAutowireCapableBeanFactory());
        jobRegistryBeanPostProcessor.afterPropertiesSet();
        return jobRegistryBeanPostProcessor;
    }

    @Bean
    public JobOperator jobOperator(){
        SimpleJobOperator simpleJobOperator=new SimpleJobOperator();
        simpleJobOperator.setJobLauncher(jobLauncher);
        simpleJobOperator.setJobParametersConverter(new DefaultJobParametersConverter());
        simpleJobOperator.setJobRegistry(jobRegistry);
        simpleJobOperator.setJobExplorer(jobExplorer);
        simpleJobOperator.setJobRepository(jobRepository);
        return simpleJobOperator;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.context=applicationContext;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

2. 编写controller

package com.it2.springbootspringbatch01.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.JobParametersInvalidException;
import org.springframework.batch.core.launch.JobInstanceAlreadyExistsException;
import org.springframework.batch.core.launch.JobOperator;
import org.springframework.batch.core.launch.NoSuchJobException;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@Slf4j
public class StartJobByJobOperatorController {

    @Autowired
    private JobOperator jobOperator;


    @RequestMapping("/jobstartbyoperator/{name}")
    public String jobstart(@PathVariable  String name) throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException, JobInstanceAlreadyExistsException, NoSuchJobException {
        log.info("通过operator启动job");
        jobOperator.start("jobDemo2","name="+name);
        return "job start by operator";
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

3. 被启动的Job,我们直接使用之前的Job

package com.it2.springbootspringbatch01.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Map;

@Configuration
//@EnableBatchProcessing
@Slf4j
public class JobParamDemo implements StepExecutionListener {

    Map<String, JobParameter> parameterMap;

    @Override
    public void beforeStep(StepExecution stepExecution) {
        parameterMap= stepExecution.getJobParameters().getParameters();
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return null;
    }

    //注入任务对象工厂
    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    //任务的执行由Step决定,注入step对象的factory
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    //创建Job对象
    @Bean
    public Job jobDemo2() {
        return jobBuilderFactory.get("jobDemo2")
                .start(step11())
                .build();

    }

    //创建Step对象

    /**
     * Job执行的Step，Job传递的数据在step中使用
     * 如何给step传递数据?
     * 使用监听器，给step传递数据,实现StepExecutionListener
     * @return
     */
    @Bean
    public Step step11() {
        return stepBuilderFactory.get("step11")
                .listener(this)
                .tasklet(new Tasklet() {
            @Override
            public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) throws Exception {
                log.info("------step11 ok");
                log.info("参数:"+parameterMap);
                return RepeatStatus.FINISHED;
            }
        }).build();
    }


}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

4. 启动服务器，发起对controller的调用，触发Job
   
   

第二节 JobOperator和JobLaucher

JobLaucher需要提前注入Job的Bean, 需要封装参数
JobOperator启动Job，只需要提供Job的Bean的名称，参数的传递直接可以使用一定格式的字符串

传送门

SpringBatch(13):JobLaucher启动Job