SpringBoot Batch

秒速五厘米 2022-07-12 04:28 124阅读 0赞

**SpringBatch主要组成**

JobRepository : 用来注册Job的容器

JobLauncher : 用来启动Job的接口

Job : 实际执行的任务，包含一个或多个Step

Step : Step 步骤，包含ItemReader、ItemProcessor、ItemWriter

ItemReader : 用来读取数据的接口

ItemProcessor : 用来处理数据的接口

ItemWriter : 用来输出数据的接口

**SpringBatch Java Config(Java 配置)**

BatchConfig.java（不完整，一部分）

@Configuration
    @EnableBatchProcessing // 开启批处理
    public class BatchConfig { 
    
        /** * 作业仓库 * * @param dataSource * @param transactionManager * @return * @throws Exception */
        @Bean
        public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception{
    
            JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
            jobRepositoryFactoryBean.setDataSource(dataSource);
            jobRepositoryFactoryBean.setTransactionManager(transactionManager);
            jobRepositoryFactoryBean.setDatabaseType(DatabaseType.MYSQL.name());
    
            return jobRepositoryFactoryBean.getObject();
        }
    
        /** * 作业调度器 * * @param dataSource * @param transactionManager * @return * @throws Exception */
        public SimpleJobLauncher jobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception{
    
            SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
            jobLauncher.setJobRepository(this.jobRepository(dataSource, transactionManager));
    
            return jobLauncher;
        }
    
        @Bean
        public Job importJob(JobBuilderFactory jobs, Step step){
            return jobs.get("importJob")
                    .incrementer(new RunIdIncrementer())
                    .flow(step)
                    .end()
                    .build();
        }
    
        @Bean
        public Step personStep(StepBuilderFactory stepBuilderFactory, 
                                    ItemReader<Person> reader, ItemWriter<Person> writer, ItemProcessor<Person, Person> processor){
            return stepBuilderFactory.get("personStep")
                    .<Person, Person>chunk(5000)
                    .reader(reader)
                    .processor(processor)
                    .writer(writer)
                    .build();
        }
    
        @Bean
        public MyJobListener myJobListener(){
            return new MyJobListener();
        }

**Job监听**

若需要监听Job的执行情况，则定义一个类实现JobExecutionListener，并在定义Job的Bean上绑定该监听器.

/** * @Title: 监听Job * @Description: * @author wangxuezheng * @date 2017年2月10日 下午2:19:12 * @version V1.0 * */
    public class MyJobListener implements JobExecutionListener{ 
    
        @Override
        public void beforeJob(JobExecution jobExecution) {
    
        }
    
        @Override
        public void afterJob(JobExecution jobExecution) {
    
        }
    
    }

@Bean
        public MyJobListener myJobListener(){
            return new MyJobListener();
        }

**数据读取**

Spring Batch为我们提供了大量的ItemReader的实现，用来读取不同的数据来源。

![这里写图片描述][SouthEast]

**数据处理及校验**

数据处理和校验都要通过ItemProcessor接口实现来完成。

(1) 数据处理

数据处理只需实现 ItemProcessor 接口，重写其 process方法。方法输入的参数是从ItemReader读取到的数据，返回的数据给ItemWriter。

public class MyItemProcessor implements ItemProcessor<Person, Person>{ 
    
        @Override
        public Person process(Person person) throws Exception {
    
            String name = person.getName().toUpperCase();
            person.setName(name);
            return person;
        }
    }

(2) 数据校验

我们可以使用JSR-303(主要实现有hibernate-validator)的注解，来校验ItemReader读取到的数据是否满足要求，ItemProcessor实现ValidationItemProcessor接口。

public class MyItemProcessor extends ValidatingItemProcessor<Person> implements ItemProcessor<Person, Person>{ 
    
        @Override
        public Person process(Person item) {
            super.process(item);
            return null;
        }
    
    }

定义校验器，实现来自于Spring的Validator接口，使用JSR-303的Validator来校验：

**数据输出**

Spring Batch 为我们提供了大量的ItemWriter的实现，用来将数据输出到不同的目的地。

![这里写图片描述][SouthEast 1]

**计划任务**

Spring Batch的任务是通过JobLauncher的run方法来执行的，因此我们只需在普通的计划任务方法中执行JobLauncher的run方法即可。

配置类使用@EnableScheduling开启计划任务支持

@Service
    public class ScheduledTaskService { 
    
        @Autowired
        JobLauncher jobLauncher;
    
        @Autowired
        Job importJob;
    
        public JobParameters jobParameters;
    
        @Scheduled(fixedRate = 5000)
        public void execute() throws Exception{
            jobParameters = new JobParametersBuilder().addLong("time", System.currentTimeMillis()).toJobParameters();
            jobLauncher.run(importJob, jobParameters);
        }
    }

**Spring Boot的支持**

SpringBoot对Spring Batch支持的源码位于org.springframework.boot.autoconfigure.batch下，SpringBoot提供了如下属性来定制SpringBatch:

spring.batch.job.names = job1,job2 #启动时要执行的Job，默认执行全部Job
    spring.batch.job.enabled=true #是否自动执行定义的Job，默认是
    spring.batch.initializer.enabled=true #是否初始化Spring Batch的数据库，默认为是
    spring.batch.schema=
    spring.batch.table-prefix= #设置SpringBatch的数据库表的前缀

[SouthEast]: /images/20220711/e3e4d21b0bcf43cb85daccb1d9f3310d.png
[SouthEast 1]: /images/20220711/32e4bcdad930468f9f9087c7f4c13cc7.png