SpringBoot Batch 实战

ゝ一纸荒年。 2022-07-10 20:12 138阅读 0赞

使用SpringBatch将csv文件中的数据使用JDBC批处理的方式插入数据库

**pom.xml依赖的jar**

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-batch</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-jdbc</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <scope>runtime</scope>
            </dependency>
    
            <dependency>
                <groupId>org.hibernate</groupId>
                <artifactId>hibernate-validator</artifactId>
            </dependency>

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

测试csv数据，位于 src/main/resources/people.csv，内容如下：

张三 18 汉族 北京  
李四 19 汉族 上海  
周杰伦 20 彝族 广州

**数据schema**

person

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

**领域模型**

Person.java

public class Person implements Serializable{ 
    
        /** * */
        private static final long serialVersionUID = 1L;
    
        private String id;
    
        @Size(max=4, min=2) // 此处使用JSR-303注解来校验数据
        private String name;
    
        private Integer age;
    
        private String nation;
    
        private String address;

**数据处理以及校验**

CsvItemProcessor.java

public class CsvItemProcessor extends ValidatingItemProcessor<Person>{ 
    
        @Override
        public Person process(Person item){
    
            super.process(item); // 需执行这句话，才会调用自定义的校验器
    
            if (item.getNation().equals("汉族")){
                item.setNation("01");
            }else{
                item.setNation("02");
            }
            return item;
        }
    }

**数据校验**

CsvBeanValidator.java

import java.util.Set;
    
    import javax.validation.ConstraintViolation;
    import javax.validation.Validation;
    import javax.validation.ValidatorFactory;
    
    import org.springframework.batch.item.validator.ValidationException;
    import org.springframework.batch.item.validator.Validator;
    import org.springframework.beans.factory.InitializingBean;
    
    public class CsvBeanValidator<T> implements Validator<T>, InitializingBean{
    
        private javax.validation.Validator validator;
    
        /** * 使用 JSR-303 的 Validator来校验我们的数据，在此处进行JSR-303的Validator的初始化 */
        @Override
        public void afterPropertiesSet() throws Exception {
            ValidatorFactory validatorFactory = Validation.buildDefaultValidatorFactory();
            validator = validatorFactory.usingContext().getValidator();
        }
    
        @Override
        public void validate(T value) throws ValidationException {
    
            // 使用Validator的validate方法校验数据
            Set<ConstraintViolation<T>> constraintViolations = validator.validate(value);
            if (constraintViolations.size() > 0){
                StringBuilder message = new StringBuilder();
                for (ConstraintViolation<T> constraintViolation : constraintViolations) {
                    message.append(constraintViolation.getMessage() + "\n");
                }
                throw new ValidationException(message.toString());
            }
        }
    }

**Job监听**

public class CsvJobListener implements JobExecutionListener{ 
    
        long startTime;
        long endTime;
    
        @Override
        public void beforeJob(JobExecution jobExecution) {
            startTime = System.currentTimeMillis();
            System.out.println("任务处理开始...");
        }
    
        @Override
        public void afterJob(JobExecution jobExecution) {
            endTime = System.currentTimeMillis();
            System.out.println("任务处理结束...");
            System.out.println("耗时:" + (endTime - startTime) + " ms");
        }
    
    }

配置类：CsvBatchConfig.java

import javax.sql.DataSource;
    
    import org.springframework.batch.core.Job;
    import org.springframework.batch.core.Step;
    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.launch.support.RunIdIncrementer;
    import org.springframework.batch.core.launch.support.SimpleJobLauncher;
    import org.springframework.batch.core.repository.JobRepository;
    import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
    import org.springframework.batch.item.ItemProcessor;
    import org.springframework.batch.item.ItemReader;
    import org.springframework.batch.item.ItemWriter;
    import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
    import org.springframework.batch.item.database.JdbcBatchItemWriter;
    import org.springframework.batch.item.file.FlatFileItemReader;
    import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
    import org.springframework.batch.item.file.mapping.DefaultLineMapper;
    import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
    import org.springframework.batch.item.validator.Validator;
    import org.springframework.batch.support.DatabaseType;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.core.io.ClassPathResource;
    import org.springframework.transaction.PlatformTransactionManager;
    
    import com.wishfulcloud.service.batch.entity.Person;
    import com.wishfulcloud.service.batch.listener.CsvJobListener;
    import com.wishfulcloud.service.batch.processor.CsvItemProcessor;
    import com.wishfulcloud.service.batch.valid.CsvBeanValidator;
    
    @Configuration
    @EnableBatchProcessing
    public class CsvBatchConfig {
    
        @Bean
        public ItemReader<Person> reader(){
            // 使用FlatFileItemReader 读取文件
            FlatFileItemReader<Person> reader = new FlatFileItemReader<Person>();
            reader.setResource(new ClassPathResource("people.csv"));
    
            reader.setLineMapper(new DefaultLineMapper<Person>(){
       {
                setLineTokenizer(new DelimitedLineTokenizer(){
       {
                    setNames(new String[]{
       "name","age","nation","address"});
                }}); 
                setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>(){
       {
                    setTargetType(Person.class);
                }});
            }});
    
            return reader;
        }
    
        @Bean
        public ItemProcessor<Person, Person> processor(){
            CsvItemProcessor processor = new CsvItemProcessor();
            processor.setValidator(csvBeanValidator());
            return processor;
        }
    
        @Bean
        public Validator<Person> csvBeanValidator(){
            return new CsvBeanValidator<Person>();
        }
    
        @Bean
        public ItemWriter<Person> writer(DataSource dataSource){
            JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<Person>();
            writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<Person>());
            String sql = "insert into person(name, age, nation, address) "
                                        + "values (:name, :age, :nation, :address)";
    
            writer.setSql(sql);
            writer.setDataSource(dataSource);
            return writer;
        }
    
        /** * 作业仓库 * * @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) // 为Job指定Step
                    .end()
                    .listener(csvJobListener()) // 绑定监听器
                    .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) // 批处理每次提交5000条数据
                    .reader(reader) // 给step绑定reader
                    .processor(processor) // 给step绑定processor
                    .writer(writer) // 给step绑定writer
                    .build();
        }
    
        @Bean
        public CsvJobListener csvJobListener(){
            return new CsvJobListener();
        }
    
    }

**运行**

启动程序，SpringBoot会自动初始化Spring Batch数据库，并将csv中的数据导入到数据库中

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

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

**项目的目录结构：**

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

[SouthEast]: /images/20220711/eeb51529958b4f7d88662265119edfbb.png
[SouthEast 1]: /images/20220711/b4d104b4b9fb484fb5bff1e0736fafe2.png
[SouthEast 2]: /images/20220711/b45dae5da38e41009cfb6893ca46ea89.png
[SouthEast 3]: /images/20220711/1b53f62924544bcc8c8e06e8287b02dc.png
[SouthEast 4]: /images/20220711/b5cada3371794ebd91f0d224b90db56a.png