ToB企服应用市场:ToB评测及商务社交产业平台

标题: Spring Boot + Spring Batch 实现批处理任务，保姆级教程！（场景实战） [打印本页]

---

作者: 民工心事    时间: 2023-8-31 22:39
标题: Spring Boot + Spring Batch 实现批处理任务，保姆级教程！（场景实战）
来源：blog.csdn.net/qq_35387940/article/details/108193473
前言

概念词就不多说了，我简单地介绍下 ， spring batch 是一个 方便使用的 较健全的 批处理 框架。
为什么说是方便使用的，因为这是 基于spring的一个框架，接入简单、易理解、流程分明。
为什么说是较健全的， 因为它提供了往常我们在对大批量数据进行处理时需要考虑到的 日志跟踪、事务粒度调配、可控执行、失败机制、重试机制、数据读写等。
正文

那么回到文章，我们该篇文章将会带来给大家的是什么？（结合实例讲解那是当然的）
从实现的业务场景来说，有以下两个：

• 从  csv文件 读取数据，进行业务处理再存储
  
• 从 数据库 读取数据，进行业务处理再存储
  

也就是平时经常遇到的数据清理或者数据过滤，又或者是数据迁移备份等等。大批量的数据，自己实现分批处理需要考虑的东西太多了，又不放心，那么使用 Spring Batch 框架 是一个很好的选择。
首先，在进入实例教程前，我们看看这次的实例里，我们使用springboot 整合spring batch 框架，要编码的东西有什么？
通过一张简单的图来了解：

可能大家看到这个图，是不是多多少少想起来定时任务框架？确实有那么点像，但是我必须在这告诉大家，这是一个批处理框架，不是一个schuedling 框架。但是前面提到它提供了可执行控制，也就是说，啥时候执行是可控的，那么显然就是自己可以进行扩展结合定时任务框架，实现你心中所想。
ok，回到主题，相信大家能从图中简单明了地看到我们这次实例，需要实现的东西有什么了。所以我就不在对各个小组件进行大批量文字的描述了。
那么我们事不宜迟，开始我们的实例教程。
首先准备一个数据库，里面建一张简单的表，用于实例数据的写入存储或者说是读取等等。
bloginfo表

相关建表sql语句：

1. CREATE TABLE `bloginfo`  (
2.   `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
3.   `blogAuthor` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '博客作者标识',
4.   `blogUrl` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '博客链接',
5.   `blogTitle` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '博客标题',
6.   `blogItem` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '博客栏目',
7.   PRIMARY KEY (`id`) USING BTREE
8. ) ENGINE = InnoDB AUTO_INCREMENT = 89031 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

复制代码

pom文件里的核心依赖：

1. <dependency>
2.     <groupId>org.springframework.boot</groupId>
3.     <artifactId>spring-boot-starter-web</artifactId>
4. </dependency>
6. <dependency>
7.     <groupId>org.springframework.boot</groupId>
8.     <artifactId>spring-boot-starter-test</artifactId>
9.     <scope>test</scope>
10. </dependency>
13. <dependency>
14.     <groupId>org.springframework.boot</groupId>
15.     <artifactId>spring-boot-starter-batch</artifactId>
16. </dependency>
19. <dependency>
20.     <groupId>org.hibernate</groupId>
21.     <artifactId>hibernate-validator</artifactId>
22.     <version>6.0.7.Final</version>
23. </dependency>
25. <dependency>
26.     <groupId>org.mybatis.spring.boot</groupId>
27.     <artifactId>mybatis-spring-boot-starter</artifactId>
28.     <version>2.0.0</version>
29. </dependency>
31. <dependency>
32.     <groupId>mysql</groupId>
33.     <artifactId>mysql-connector-java</artifactId>
34.     <scope>runtime</scope>
35. </dependency>
38. <dependency>
39.     <groupId>com.alibaba</groupId>
40.     <artifactId>druid-spring-boot-starter</artifactId>
41.     <version>1.1.18</version>
42. </dependency>

复制代码

yml文件：
Spring Boot 基础就不介绍了，推荐看这个实战项目：

https://github.com/javastacks/spring-boot-best-practice

1. spring:
2.   batch:
3.     job:
4. #设置为 false -需要jobLaucher.run执行
5.       enabled: false
6.     initialize-schema: always
7. #    table-prefix: my-batch
9.   datasource:
10.     druid:
11.       username: root
12.       password: root
13.       url: jdbc:mysql://localhost:3306/hellodemo?useSSL=false&useUnicode=true&characterEncoding=UTF-8&serverTimezone=GMT%2B8&zeroDateTimeBehavior=convertToNull
14.       driver-class-name: com.mysql.cj.jdbc.Driver
15.       initialSize: 5
16.       minIdle: 5
17.       maxActive: 20
18.       maxWait: 60000
19.       timeBetweenEvictionRunsMillis: 60000
20.       minEvictableIdleTimeMillis: 300000
21.       validationQuery: SELECT 1 FROM DUAL
22.       testWhileIdle: true
23.       testOnBorrow: false
24.       testOnReturn: false
25.       poolPreparedStatements: true
26.       maxPoolPreparedStatementPerConnectionSize: 20
27.       useGlobalDataSourceStat: true
28.       connectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000
29. server:
30.   port: 8665

复制代码

ps：这里我们用到了druid数据库连接池，其实有个小坑，后面文章会讲到。

因为我们这次的实例最终数据处理完之后，是写入数据库存储（当然你也可以输出到文件等等）。
所以我们前面也建了一张表，pom文件里面我们也整合的mybatis，那么我们在整合spring batch 主要编码前，我们先把这些关于数据库打通用到的简单过一下。
pojo 层

BlogInfo.java ：

1. /**
2. * @Author : JCccc
3. * @Description :
4. **/
5. public class BlogInfo {
7.     private Integer id;
8.     private String blogAuthor;
9.     private String blogUrl;
10.     private String blogTitle;
11.     private String blogItem;
13.     @Override
14.     public String toString() {
15.         return "BlogInfo{" +
16.                 "id=" + id +
17.                 ", blogAuthor='" + blogAuthor + '\'' +
18.                 ", blogUrl='" + blogUrl + '\'' +
19.                 ", blogTitle='" + blogTitle + '\'' +
20.                 ", blogItem='" + blogItem + '\'' +
21.                 '}';
22.     }
24.     public Integer getId() {
25.         return id;
26.     }
28.     public void setId(Integer id) {
29.         this.id = id;
30.     }
32.     public String getBlogAuthor() {
33.         return blogAuthor;
34.     }
36.     public void setBlogAuthor(String blogAuthor) {
37.         this.blogAuthor = blogAuthor;
38.     }
40.     public String getBlogUrl() {
41.         return blogUrl;
42.     }
44.     public void setBlogUrl(String blogUrl) {
45.         this.blogUrl = blogUrl;
46.     }
48.     public String getBlogTitle() {
49.         return blogTitle;
50.     }
52.     public void setBlogTitle(String blogTitle) {
53.         this.blogTitle = blogTitle;
54.     }
56.     public String getBlogItem() {
57.         return blogItem;
58.     }
60.     public void setBlogItem(String blogItem) {
61.         this.blogItem = blogItem;
62.     }
63. }

复制代码

mapper层

BlogMapper.java ：

ps：可以看到这个实例我用的是注解的方式，哈哈为了省事，而且我还不写servcie层和impl层，也是为了省事，因为该篇文章重点不在这些，所以这些不好的大家不要学。

1. import com.example.batchdemo.pojo.BlogInfo;
2. import org.apache.ibatis.annotations.*;
3. import java.util.List;
4. import java.util.Map;
6. /**
7. * @Author : JCccc
8. * @Description :
9. **/
10. @Mapper
11. public interface BlogMapper {
12.     @Insert("INSERT INTO bloginfo ( blogAuthor, blogUrl, blogTitle, blogItem )   VALUES ( #{blogAuthor}, #{blogUrl},#{blogTitle},#{blogItem}) ")
13.     @Options(useGeneratedKeys = true, keyProperty = "id")
14.     int insert(BlogInfo bloginfo);
16.     @Select("select blogAuthor, blogUrl, blogTitle, blogItem from bloginfo where blogAuthor < #{authorId}")
17.      List<BlogInfo> queryInfoById(Map<String , Integer> map);
19. }

复制代码

接下来 ，重头戏，我们开始对前边那张图里涉及到的各个小组件进行编码。
首先创建一个 配置类， MyBatchConfig.java：
从我起名来看，可以知道这基本就是咱们整合spring batch 涉及到的一些配置组件都会写在这里了。
首先我们按照咱们上面的图来看，里面包含内容有：

1. JobRepository job的注册/存储器
2. JobLauncher job的执行器
3. Job job任务，包含一个或多个Step
4. Step 包含（ItemReader、ItemProcessor和ItemWriter)
5. ItemReader 数据读取器
6. ItemProcessor 数据处理器
7. ItemWriter 数据输出器

复制代码

首先，在MyBatchConfig类前加入注解：
@Configuration 用于告诉spring，咱们这个类是一个自定义配置类，里面很多bean都需要加载到spring容器里面
@EnableBatchProcessing 开启批处理支持

然后开始往MyBatchConfig类里，编写各个小组件。
JobRepository

写在MyBatchConfig类里

1. /**
2. * JobRepository定义：Job的注册容器以及和数据库打交道（事务管理等）
3. * @param dataSource
4. * @param transactionManager
5. * @return
6. * @throws Exception
7. */
8. @Bean
9. public JobRepository myJobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception{
10.     JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
11.     jobRepositoryFactoryBean.setDatabaseType("mysql");
12.     jobRepositoryFactoryBean.setTransactionManager(transactionManager);
13.     jobRepositoryFactoryBean.setDataSource(dataSource);
14.     return jobRepositoryFactoryBean.getObject();
15. }

复制代码

JobLauncher

写在MyBatchConfig类里

1. /**
2. * jobLauncher定义：job的启动器,绑定相关的jobRepository
3. * @param dataSource
4. * @param transactionManager
5. * @return
6. * @throws Exception
7. */
8. @Bean
9. public SimpleJobLauncher myJobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception{
10.     SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
11.     // 设置jobRepository
12.     jobLauncher.setJobRepository(myJobRepository(dataSource, transactionManager));
13.     return jobLauncher;
14. }

复制代码

Job

写在MyBatchConfig类里

1. /**
2. * 定义job
3. * @param jobs
4. * @param myStep
5. * @return
6. */
7. @Bean
8. public Job myJob(JobBuilderFactory jobs, Step myStep){
9.     return jobs.get("myJob")
10.             .incrementer(new RunIdIncrementer())
11.             .flow(myStep)
12.             .end()
13.             .listener(myJobListener())
14.             .build();
15. }

复制代码

对于Job的运行，是可以配置监听器的
JobListener

写在MyBatchConfig类里

1. /**
2. * 注册job监听器
3. * @return
4. */
5. @Bean
6. public MyJobListener myJobListener(){
7.     return new MyJobListener();
8. }

复制代码

这是一个我们自己自定义的监听器，所以是单独创建的，MyJobListener.java：

1. /**
2. * @Author : JCccc
3. * @Description :监听Job执行情况，实现JobExecutorListener，且在batch配置类里，Job的Bean上绑定该监听器
4. **/
6. public class MyJobListener implements JobExecutionListener {
8.     private Logger logger = LoggerFactory.getLogger(MyJobListener.class);
10.     @Override
11.     public void beforeJob(JobExecution jobExecution) {
12.         logger.info("job 开始, id={}",jobExecution.getJobId());
13.     }
15.     @Override
16.     public void afterJob(JobExecution jobExecution) {
17.         logger.info("job 结束, id={}",jobExecution.getJobId());
18.     }
19. }

复制代码

Step（ItemReader  ItemProcessor  ItemWriter）

step里面包含数据读取器，数据处理器，数据输出器三个小组件的的实现。
我们也是一个个拆解来进行编写。
文章前边说到，该篇实现的场景包含两种，一种是从csv文件读入大量数据进行处理，另一种是从数据库表读入大量数据进行处理。
从CSV文件读取数据

ItemReader

写在MyBatchConfig类里

1. /**
2. * ItemReader定义：读取文件数据+entirty实体类映射
3. * @return
4. */
5. @Bean
6. public ItemReader<BlogInfo> reader(){
7.     // 使用FlatFileItemReader去读cvs文件，一行即一条数据
8.     FlatFileItemReader<BlogInfo> reader = new FlatFileItemReader<>();
9.     // 设置文件处在路径
10.     reader.setResource(new ClassPathResource("static/bloginfo.csv"));
11.     // entity与csv数据做映射
12.     reader.setLineMapper(new DefaultLineMapper<BlogInfo>() {
13.         {
14.             setLineTokenizer(new DelimitedLineTokenizer() {
15.                 {
16.                     setNames(new String[]{"blogAuthor","blogUrl","blogTitle","blogItem"});
17.                 }
18.             });
19.             setFieldSetMapper(new BeanWrapperFieldSetMapper<BlogInfo>() {
20.                 {
21.                     setTargetType(BlogInfo.class);
22.                 }
23.             });
24.         }
25.     });
26.     return reader;
27. }

复制代码

简单代码解析：

对于数据读取器 ItemReader ，我们给它安排了一个读取监听器，创建 MyReadListener.java ：

1. /**
2. * @Author : JCccc
3. * @Description :
4. **/
6. public class MyReadListener implements ItemReadListener<BlogInfo> {
8.     private Logger logger = LoggerFactory.getLogger(MyReadListener.class);
10.     @Override
11.     public void beforeRead() {
12.     }
14.     @Override
15.     public void afterRead(BlogInfo item) {
16.     }
18.     @Override
19.     public void onReadError(Exception ex) {
20.         try {
21.             logger.info(format("%s%n", ex.getMessage()));
22.         } catch (Exception e) {
23.             e.printStackTrace();
24.         }
25.     }
26. }

复制代码

ItemProcessor

写在MyBatchConfig类里

1. /**
2. * 注册ItemProcessor: 处理数据+校验数据
3. * @return
4. */
5. @Bean
6. public ItemProcessor<BlogInfo, BlogInfo> processor(){
7.     MyItemProcessor myItemProcessor = new MyItemProcessor();
8.     // 设置校验器
9.     myItemProcessor.setValidator(myBeanValidator());
10.     return myItemProcessor;
11. }

复制代码

数据处理器，是我们自定义的，里面主要是包含我们对数据处理的业务逻辑，并且我们设置了一些数据校验器，我们这里使用 JSR-303的Validator来作为校验器。
校验器

写在MyBatchConfig类里

1. /**
2. * 注册校验器
3. * @return
4. */
5. @Bean
6. public MyBeanValidator myBeanValidator(){
7.     return new MyBeanValidator<BlogInfo>();
8. }

复制代码

创建MyItemProcessor.java ：

ps：里面我的数据处理逻辑是，获取出读取数据里面的每条数据的blogItem字段，如果是springboot，那就对title字段值进行替换。

其实也就是模拟一个简单地数据处理场景。

1. import com.example.batchdemo.pojo.BlogInfo;
2. import org.springframework.batch.item.validator.ValidatingItemProcessor;
3. import org.springframework.batch.item.validator.ValidationException;
5. /**
6. * @Author : JCccc
7. * @Description :
8. **/
9. public class MyItemProcessor extends ValidatingItemProcessor<BlogInfo> {
10.     @Override
11.     public BlogInfo process(BlogInfo item) throws ValidationException {
12.         /**
13.          * 需要执行super.process(item)才会调用自定义校验器
14.          */
15.         super.process(item);
16.         /**
17.          * 对数据进行简单的处理
18.          */
19.         if (item.getBlogItem().equals("springboot")) {
20.             item.setBlogTitle("springboot 系列还请看看我Jc");
21.         } else {
22.             item.setBlogTitle("未知系列");
23.         }
24.         return item;
25.     }
26. }

复制代码

创建MyBeanValidator.java：

1. import org.springframework.batch.item.validator.ValidationException;
2. import org.springframework.batch.item.validator.Validator;
3. import org.springframework.beans.factory.InitializingBean;
4. import javax.validation.ConstraintViolation;
5. import javax.validation.Validation;
6. import javax.validation.ValidatorFactory;
7. import java.util.Set;
9. /**
10. * @Author : JCccc
11. * @Description :
12. **/
13. public class MyBeanValidator<T> implements Validator<T>, InitializingBean {
15.     private javax.validation.Validator validator;
17.     @Override
18.     public void validate(T value) throws ValidationException {
19.         /**
20.          * 使用Validator的validate方法校验数据
21.          */
22.         Set<ConstraintViolation<T>> constraintViolations =
23.                 validator.validate(value);
24.         if (constraintViolations.size() > 0) {
25.             StringBuilder message = new StringBuilder();
26.             for (ConstraintViolation<T> constraintViolation : constraintViolations) {
27.                 message.append(constraintViolation.getMessage() + "\n");
28.             }
29.             throw new ValidationException(message.toString());
30.         }
31.     }
33.     /**
34.      * 使用JSR-303的Validator来校验我们的数据，在此进行JSR-303的Validator的初始化
35.      * @throws Exception
36.      */
37.     @Override
38.     public void afterPropertiesSet() throws Exception {
39.         ValidatorFactory validatorFactory =
40.                 Validation.buildDefaultValidatorFactory();
41.         validator = validatorFactory.usingContext().getValidator();
42.     }
44. }

复制代码

ps：其实该篇文章没有使用这个数据校验器，大家想使用的话，可以在实体类上添加一些校验器的注解@NotNull @Max @Email等等。我偏向于直接在处理器里面进行处理，想把关于数据处理的代码都写在一块。

ItemWriter

写在MyBatchConfig类里

1. /**
2. * ItemWriter定义：指定datasource，设置批量插入sql语句，写入数据库
3. * @param dataSource
4. * @return
5. */
6. @Bean
7. public ItemWriter<BlogInfo> writer(DataSource dataSource){
8.     // 使用jdbcBcatchItemWrite写数据到数据库中
9.     JdbcBatchItemWriter<BlogInfo> writer = new JdbcBatchItemWriter<>();
10.     // 设置有参数的sql语句
11.     writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<BlogInfo>());
12.     String sql = "insert into bloginfo "+" (blogAuthor,blogUrl,blogTitle,blogItem) "
13.             +" values(:blogAuthor,:blogUrl,:blogTitle,:blogItem)";
14.     writer.setSql(sql);
15.     writer.setDataSource(dataSource);
16.     return writer;
17. }

复制代码

简单代码解析：

同样 对于数据读取器 ItemWriter ，我们给它也安排了一个输出监听器，创建 MyWriteListener.java：
[code]import com.example.batchdemo.pojo.BlogInfo;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.batch.core.ItemWriteListener;import java.util.List;import static java.lang.String.format;/** * @Author : JCccc * @Description : **/public class MyWriteListener implements ItemWriteListener {    private Logger logger = LoggerFactory.getLogger(MyWriteListener.class);    @Override    public void beforeWrite(List

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4