Spring Boot整合Spring Batch

快来打我* 2022-02-03 15:34 462阅读 0赞

引言

　　Spring Batch是处理大量数据操作的一个框架，主要用来读取大量数据，然后进行一定的处理后输出指定的形式。比如我们可以将csv文件中的数据（数据量几百万甚至几千万都是没问题的）批处理插入保存到数据库中，就可以使用该框架，但是不管是数据资料还是网上资料，我看到很少有这样的详细讲解。所以本篇文章的主要目的边讲解的同时边实战（其中的代码都是经过实践的）。同样地先从Spring Boot对Batch框架的支持说起，最后一步一步进行代码实践！

一、Spring Boot对Batch框架的支持

1、Spring Batch框架的组成部分

　　1）JobRepository：用来注册Job容器，设置数据库相关属性。

　　2）JobLauncher：用来启动Job的接口

　　3）Job：我们要实际执行的任务，包含一个或多个

　　4）Step：即步骤，包括：ItemReader->ItemProcessor->ItemWriter

　　5）ItemReader：用来读取数据，做实体类与数据字段之间的映射。比如读取csv文件中的人员数据，之后对应实体person的字段做mapper

　　6）ItemProcessor：用来处理数据的接口，同时可以做数据校验（设置校验器，使用JSR-303(hibernate-validator)注解），比如将中文性别男/女，转为M/F。同时校验年龄字段是否符合要求等

　　7）ItemWriter：用来输出数据的接口，设置数据库源。编写预处理SQL插入语句

以上七个组成部分，只需要在配置类中逐一注册即可，同时配置类需要开启@EnableBatchProcessing注解

  1. @Configuration
  2. @EnableBatchProcessing // 开启批处理的支持
  3. @Import(DruidDBConfig.class) // 注入datasource
  4. public class CsvBatchConfig {
  6. }

2、批处理流程图

如下流程图即可以解释在配置类中为什么需要这么定义，具体请看实战部分的代码。

1352849-20190509145057653-369261159.png

二、实战

1、添加依赖

1）spring batch依赖

  1. <!--  spring batch -->
  2. <dependency>
  3.     <groupId>org.springframework.boot</groupId>
  4.     <artifactId>spring-boot-starter-batch</artifactId>
  5. </dependency>

2）校验器依赖

  1. <!-- hibernate validator -->
  2. <dependency>
  3.     <groupId>org.hibernate</groupId>
  4.     <artifactId>hibernate-validator</artifactId>
  5.     <version>6.0.7.Final</version>
  6. </dependency>

3）mysql+druid依赖

  1. <!-- mysql connector-->
  2. <dependency>
  3.     <groupId>mysql</groupId>
  4.     <artifactId>mysql-connector-java</artifactId>
  5.     <version>5.1.35</version>
  6. </dependency>
  7. <!-- alibaba dataSource -->
  8. <dependency>
  9.     <groupId>com.alibaba</groupId>
  10.     <artifactId>druid</artifactId>
  11.     <version>1.1.12</version>
  12. </dependency>

4）test测试依赖

  1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-test</artifactId>
  4. </dependency>

2、application.yml配置

当job发布开始执行任务时，spring batch会自动生成相关的batch开头的表。这些表一开始是不存在的！需要在application配置文件中做相关的设置。

1352849-20190509150408907-104253959.png

  1. # batch
  2.   batch:
  3.     job:
  4.       # 默认自动执行定义的Job(true)，改为false，需要jobLaucher.run执行
  5.       enabled: false
  6.     # spring batch在数据库里面创建默认的数据表，如果不是always则会提示相关表不存在
  7.     initialize-schema: always
  8.     # 设置batch表的前缀
  9. #    table-prefix: csv-batch

3、数据源配置

  1. datasource:
  2.     username: root
  3.     password: 1234
  4.     url: jdbc:mysql://127.0.0.1:3306/db_base?useSSL=false&serverTimezone=UTC&characterEncoding=utf8
  5.     driver-class-name: com.mysql.jdbc.Driver

注册DBConfig配置类：之后通过import导入batch配置类中

ContractedBlock.gif

4、编写batch配置类

在配置类中，注册Spring Batch的各个组成部分即可，其中部分说明已在代码中注释.

ContractedBlock.gif

5、定义处理器

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

  1. /**
  2.  * @author jian
  3.  * @date 2019/4/28
  4.  * @description
  5.  * CSV文件数据处理及校验
  6.  * 只需要实现ItemProcessor接口，重写process方法，输入的参数是从ItemReader读取到的数据，返回的数据给ItemWriter
  7.  */
  8. public class CvsItemProcessor extends ValidatingItemProcessor<Person> {
  9.     private Logger logger = LoggerFactory.getLogger(CvsItemProcessor.class);
  11.     @Override
  12.     public Person process(Person item) throws ValidationException {
  13.         // 执行super.process()才能调用自定义的校验器
  14.         logger.info("processor start validating...");
  15.         super.process(item);
  17.         // 数据处理，比如将中文性别设置为M/F
  18.         if ("男".equals(item.getGender())) {
  19.             item.setGender("M");
  20.         } else {
  21.             item.setGender("F");
  22.         }
  23.         logger.info("processor end validating...");
  24.         return item;
  25.     }
  26. }

6、定义校验器

定义校验器：使用JSR-303(hibernate-validator)注解，来校验ItemReader读取到的数据是否满足要求。如不满足则不会进行接下来的批处理任务。

ContractedBlock.gif

7、定义监听器：

监听Job执行情况，则定义一个类实现JobExecutorListener，并定义Job的Bean上绑定该监听器

  1. /**
  2.  * @author jian
  3.  * @date 2019/4/28
  4.  * @description
  5.  * 监听Job执行情况，则定义一个类实现JobExecutorListener，并定义Job的Bean上绑定该监听器
  6.  */
  7. public class CsvJobListener implements JobExecutionListener {
  9.     private Logger logger = LoggerFactory.getLogger(CsvJobListener.class);
  10.     private long startTime;
  11.     private long endTime;
  13.     @Override
  14.     public void beforeJob(JobExecution jobExecution) {
  15.         startTime = System.currentTimeMillis();
  16.         logger.info("job process start...");
  17.     }
  19.     @Override
  20.     public void afterJob(JobExecution jobExecution) {
  21.         endTime = System.currentTimeMillis();
  22.         logger.info("job process end...");
  23.         logger.info("elapsed time: " + (endTime - startTime) + "ms");
  24.     }
  25. }

三、测试

1、person.csv文件

csv文件时以逗号为分隔的数据表示字段，回车表示一行（条）数据记录

  1. 1,Zhangsan,21,男
  2. 2,Lisi,22,女
  3. 3,Wangwu,23,男
  4. 4,Zhaoliu,24,男
  5. 5,Zhouqi,25,女

放在resources下，在ItemReader中读取的该路径即可

1352849-20190509151040677-216373897.png

2、person实体

person.csv中的字段与之对应，并在该实体中可以添加校验注解，如@Size表示该字段的长度范围，如果超过规定。则会被校验检测到，批处理将不会进行！

ContractedBlock.gif

3、数据表

  1. CREATE TABLE `person` (
  2.   `id` int(11) NOT NULL,
  3.   `name` varchar(10) DEFAULT NULL,
  4.   `age` int(11) DEFAULT NULL,
  5.   `gender` varchar(2) NOT NULL,
  6.   PRIMARY KEY (`id`)
  7. ) ENGINE=InnoDB DEFAULT CHARSET=latin1

一开始表是没有数据的

1352849-20190509151947698-983575510.png

4、测试类

需要注入发布器，与job任务。同时可以使用后置参数灵活处理，最后调用JobLauncher.run方法执行批处理任务

  1. @RunWith(SpringRunner.class)
  2. @SpringBootTest
  3. public class BatchTest {
  5.     @Autowired
  6.     SimpleJobLauncher jobLauncher;
  8.     @Autowired
  9.     Job importJob;
  11.     @Test
  12.     public void test() throws Exception{
  13.         // 后置参数：使用JobParameters中绑定参数
  14.         JobParameters jobParameters = new JobParametersBuilder().addLong("time", System.currentTimeMillis())
  15.                 .toJobParameters();
  16.         jobLauncher.run(importJob, jobParameters);
  17.     }
  18. }

5、测试结果

  1. ....
  2. 2019-05-09 15:23:39.576  INFO 18296 --- [           main] com.lijian.test.BatchTest                : Started BatchTest in 6.214 seconds (JVM running for 7.185)
  3. 2019-05-09 15:23:39.939  INFO 18296 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=importCsvJob]] launched with the following parameters: [{time=1557386619763}]
  4. 2019-05-09 15:23:39.982  INFO 18296 --- [           main] com.lijian.config.batch.CsvJobListener   : job process start...
  5. 2019-05-09 15:23:40.048  INFO 18296 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step]
  6. 2019-05-09 15:23:40.214  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
  7. 2019-05-09 15:23:40.282  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor end validating...
  8. 2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
  9. 2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor end validating...
  10. 2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
  11. 2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor end validating...
  12. 2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
  13. 2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor end validating...
  14. 2019-05-09 15:23:40.283  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
  15. 2019-05-09 15:23:40.284  INFO 18296 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor end validating...
  16. 2019-05-09 15:23:40.525  INFO 18296 --- [           main] com.lijian.config.batch.CsvJobListener   : job process end...
  17. 2019-05-09 15:23:40.526  INFO 18296 --- [           main] com.lijian.config.batch.CsvJobListener   : elapsed time: 543ms
  18. 2019-05-09 15:23:40.548  INFO 18296 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=importCsvJob]] completed with the following parameters: [{time=1557386619763}] and the following status: [COMPLETED]
  19. 2019-05-09 15:23:40.564  INFO 18296 --- [       Thread-5] com.alibaba.druid.pool.DruidDataSource   : {dataSource-1} closed

查看表中数据： select * from person;

1352849-20190509152601155-880594872.png

若继续插入数据，并且测试校验器是否生效，则将person.csv更改为如下内容：

  1. 6,springbatch,24,男
  2. 7,springboot,23,女

由于实体类中JSR校验注解对name长度范围进行了检验，即添加了 @Size(min=2, max=8) 的注解。故会报错显示校验不通过，批处理将不会进行。

  1. ...
  2. Started BatchTest in 5.494 seconds (JVM running for 6.41)
  3. 2019-05-09 15:30:02.147  INFO 20368 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=importCsvJob]] launched with the following parameters: [{time=1557387001499}]
  4. 2019-05-09 15:30:02.247  INFO 20368 --- [           main] com.lijian.config.batch.CsvJobListener   : job process start...
  5. 2019-05-09 15:30:02.503  INFO 20368 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [step]
  6. 2019-05-09 15:30:02.683  INFO 20368 --- [           main] c.lijian.config.batch.CvsItemProcessor   : processor start validating...
  7. 2019-05-09 15:30:02.761 ERROR 20368 --- [           main] o.s.batch.core.step.AbstractStep         : Encountered an error executing step step in job importCsvJob
  9. org.springframework.batch.item.validator.ValidationException: size must be between 2 and 8
  10. ...

发表评论

表情：
评论列表 （有 0 条评论，462人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Spring Boot整合Spring Batch

    引言 > 　　Spring Batch是处理大量数据操作的一个框架，主要用来读取大量数据，然后进行一定的处理后输出指定的形式。比如我们可以将csv文件中的数据（数据量几百

    快来打我*快来打我*/ 0 赞/ 463 阅读