Spring Boot + Spring Batch 实现批处理任务，保姆级教程！

前言

概念词就不多说了，我简单地介绍下 ， spring batch 是一个 方便使用的 较健全的 批处理 框架。

为什么说是方便使用的，因为这是 基于spring的一个框架，接入简单、易理解、流程分明。

为什么说是较健全的， 因为它提供了往常我们在对大批量数据进行处理时需要考虑到的 日志跟踪、事务粒度调配、可控执行、失败机制、重试机制、数据读写等。

正文

那么回到文章，我们该篇文章将会带来给大家的是什么？（结合实例讲解那是当然的）

从实现的业务场景来说，有以下两个：

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

也就是平时经常遇到的数据清理或者数据过滤，又或者是数据迁移备份等等。大批量的数据，自己实现分批处理需要考虑的东西太多了，又不放心，那么使用 Spring Batch 框架 是一个很好的选择。

首先，在进入实例教程前，我们看看这次的实例里，我们使用springboot 整合spring batch 框架，要编码的东西有什么？

通过一张简单的图来了解：

可能大家看到这个图，是不是多多少少想起来定时任务框架？确实有那么点像，但是我必须在这告诉大家，这是一个批处理框架，不是一个schuedling 框架。但是前面提到它提供了可执行控制，也就是说，啥时候执行是可控的，那么显然就是自己可以进行扩展结合定时任务框架，实现你心中所想。

ok，回到主题，相信大家能从图中简单明了地看到我们这次实例，需要实现的东西有什么了。所以我就不在对各个小组件进行大批量文字的描述了。

那么我们事不宜迟，开始我们的实例教程。

首先准备一个数据库，里面建一张简单的表，用于实例数据的写入存储或者说是读取等等。

bloginfo表

相关建表sql语句：

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

pom文件里的核心依赖：

    org.springframework.boot
    spring-boot-starter-web


    org.springframework.boot
    spring-boot-starter-test
    test


    org.springframework.boot
    spring-boot-starter-batch


    org.hibernate
    hibernate-validator
    6.0.7.Final


    org.mybatis.spring.boot
    mybatis-spring-boot-starter
    2.0.0


    mysql
    mysql-connector-java
    runtime


    com.alibaba
    druid-spring-boot-starter
    1.1.18

yml文件：

spring:
  batch:
    job:
#设置为 false -需要jobLaucher.run执行
      enabled: false
    initialize-schema: always
#    table-prefix: my-batch
  datasource:
    druid:
      username: root
      password: root
      url: jdbc:mysql://localhost:3306/hellodemo?useSSL=false&useUnicode=true&characterEncoding=UTF-8&serverTimezone=GMT%2B8&zeroDateTimeBehavior=convertToNull
      driver-class-name: com.mysql.cj.jdbc.Driver
      initialSize: 5
      minIdle: 5
      maxActive: 20
      maxWait: 60000
      timeBetweenEvictionRunsMillis: 60000
      minEvictableIdleTimeMillis: 300000
      validationQuery: SELECT 1 FROM DUAL
      testWhileIdle: true
      testOnBorrow: false
      testOnReturn: false
      poolPreparedStatements: true
      maxPoolPreparedStatementPerConnectionSize: 20
      useGlobalDataSourceStat: true
      connectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000
server:
  port: 8665

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

因为我们这次的实例最终数据处理完之后，是写入数据库存储（当然你也可以输出到文件等等）。

所以我们前面也建了一张表，pom文件里面我们也整合的mybatis，那么我们在整合spring batch 主要编码前，我们先把这些关于数据库打通用到的简单过一下。

pojo 层

BlogInfo.java ：

@Data
public class BlogInfo {
    private Integer id;
    private String blogAuthor;
    private String blogUrl;
    private String blogTitle;
    private String blogItem;
    @Override
    public String toString() {
        return "BlogInfo{" +
                "id=" + id +
                ", blogAuthor='" + blogAuthor + '\'' +
                ", blogUrl='" + blogUrl + '\'' +
                ", blogTitle='" + blogTitle + '\'' +
                ", blogItem='" + blogItem + '\'' +
                '}';
    }
 }

mapper层

BlogMapper.java ：

ps：可以看到这个实例我用的是注解的方式，哈哈为了省事，servcie层和impl层，就省略不写了。

import com.example.batchdemo.pojo.BlogInfo;
import org.apache.ibatis.annotations.*;
import java.util.List;
import java.util.Map;
@Mapper
public interface BlogMapper {
    @Insert("INSERT INTO bloginfo ( blogAuthor, blogUrl, blogTitle, blogItem )   VALUES ( #{blogAuthor}, #{blogUrl},#{blogTitle},#{blogItem}) ")
    @Options(useGeneratedKeys = true, keyProperty = "id")
    int insert(BlogInfo bloginfo);
    @Select("select blogAuthor, blogUrl, blogTitle, blogItem from bloginfo where blogAuthor  
接下来 ，重头戏，我们开始对前边那张图里涉及到的各个小组件进行编码。
 
首先创建一个 配置类， MyBatchConfig.java：
 
从我起名来看，可以知道这基本就是咱们整合spring batch 涉及到的一些配置组件都会写在这里了。
 
首先我们按照咱们上面的图来看，里面包含内容有：
 
JobRepository job的注册/存储器
JobLauncher job的执行器
Job job任务，包含一个或多个Step
Step 包含（ItemReader、ItemProcessor和ItemWriter)
ItemReader 数据读取器
ItemProcessor 数据处理器
ItemWriter 数据输出器
 
首先，在MyBatchConfig类前加入注解：
 
@Configuration 用于告诉spring，咱们这个类是一个自定义配置类，里面很多bean都需要加载到spring容器里面
 
@EnableBatchProcessing 开启批处理支持
然后开始往MyBatchConfig类里，编写各个小组件。
 
JobRepository
写在MyBatchConfig类里
 
/**
 * JobRepository定义：Job的注册容器以及和数据库打交道（事务管理等）
 * @param dataSource
 * @param transactionManager
 * @return
 * @throws Exception
 */
@Bean
public JobRepository myJobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception{
    JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
    jobRepositoryFactoryBean.setDatabaseType("mysql");
    jobRepositoryFactoryBean.setTransactionManager(transactionManager);
    jobRepositoryFactoryBean.setDataSource(dataSource);
    return jobRepositoryFactoryBean.getObject();
}
 
JobLauncher
写在MyBatchConfig类里
 
/**
 * jobLauncher定义：job的启动器,绑定相关的jobRepository
 * @param dataSource
 * @param transactionManager
 * @return
 * @throws Exception
 */
@Bean
public SimpleJobLauncher myJobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception{
    SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
    // 设置jobRepository
    jobLauncher.setJobRepository(myJobRepository(dataSource, transactionManager));
    return jobLauncher;
}
 
Job
写在MyBatchConfig类里
 
/**
 * 定义job
 * @param jobs
 * @param myStep
 * @return
 */
@Bean
public Job myJob(JobBuilderFactory jobs, Step myStep){
    return jobs.get("myJob")
            .incrementer(new RunIdIncrementer())
            .flow(myStep)
            .end()
            .listener(myJobListener())
            .build();
}
 
对于Job的运行，是可以配置监听器的
 
JobListener
写在MyBatchConfig类里
 
/**
 * 注册job监听器
 * @return
 */
@Bean
public MyJobListener myJobListener(){
    return new MyJobListener();
}
 
这是一个我们自己自定义的监听器，所以是单独创建的，MyJobListener.java：
 
/**
 * @Author : JCccc
 * @Description :监听Job执行情况，实现JobExecutorListener，且在batch配置类里，Job的Bean上绑定该监听器
 **/
public class MyJobListener implements JobExecutionListener {
    private Logger logger = LoggerFactory.getLogger(MyJobListener.class);
    @Override
    public void beforeJob(JobExecution jobExecution) {
        logger.info("job 开始, id={}",jobExecution.getJobId());
    }
    @Override
    public void afterJob(JobExecution jobExecution) {
        logger.info("job 结束, id={}",jobExecution.getJobId());
    }
}
 
Step（ItemReader ItemProcessor ItemWriter）
step里面包含数据读取器，数据处理器，数据输出器三个小组件的的实现。
 
我们也是一个个拆解来进行编写。
 
文章前边说到，该篇实现的场景包含两种，一种是从csv文件读入大量数据进行处理，另一种是从数据库表读入大量数据进行处理。
 
从CSV文件读取数据
ItemReader
写在MyBatchConfig类里
 
/**
 * ItemReader定义：读取文件数据+entirty实体类映射
 * @return
 */
@Bean
public ItemReader reader(){
    // 使用FlatFileItemReader去读cvs文件，一行即一条数据
    FlatFileItemReader reader = new FlatFileItemReader();
    // 设置文件处在路径
    reader.setResource(new ClassPathResource("static/bloginfo.csv"));
    // entity与csv数据做映射
    reader.setLineMapper(new DefaultLineMapper() {
        {
            setLineTokenizer(new DelimitedLineTokenizer() {
                {
                    setNames(new String[]{"blogAuthor","blogUrl","blogTitle","blogItem"});
                }
            });
            setFieldSetMapper(new BeanWrapperFieldSetMapper() {
                {
                    setTargetType(BlogInfo.class);
                }
            });
        }
    });
    return reader;
}
 
简单代码解析：
对于数据读取器 ItemReader ，我们给它安排了一个读取监听器，创建 MyReadListener.java ：
 
public class MyReadListener implements ItemReadListener {
    private Logger logger = LoggerFactory.getLogger(MyReadListener.class);
    @Override
    public void beforeRead() {
    }
    @Override
    public void afterRead(BlogInfo item) {
    }
    @Override
    public void onReadError(Exception ex) {
        try {
            logger.info(format("%s%n", ex.getMessage()));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
 
ItemProcessor
写在MyBatchConfig类里
 
/**
 * 注册ItemProcessor: 处理数据+校验数据
 * @return
 */
@Bean
public ItemProcessor processor(){
    MyItemProcessor myItemProcessor = new MyItemProcessor();
    // 设置校验器
    myItemProcessor.setValidator(myBeanValidator());
    return myItemProcessor;
}
 
数据处理器，是我们自定义的，里面主要是包含我们对数据处理的业务逻辑，并且我们设置了一些数据校验器，我们这里使用 JSR-303的Validator来作为校验器。
 
校验器
写在MyBatchConfig类里
 
/**
 * 注册校验器
 * @return
 */
@Bean
public MyBeanValidator myBeanValidator(){
    return new MyBeanValidator();
}
 
创建MyItemProcessor.java ：
 
ps：里面我的数据处理逻辑是，获取出读取数据里面的每条数据的blogItem字段，如果是springboot，那就对title字段值进行替换。
 
其实也就是模拟一个简单地数据处理场景。
 
import com.example.batchdemo.pojo.BlogInfo;
import org.springframework.batch.item.validator.ValidatingItemProcessor;
import org.springframework.batch.item.validator.ValidationException;
public class MyItemProcessor extends ValidatingItemProcessor {
    @Override
    public BlogInfo process(BlogInfo item) throws ValidationException {
        /**
         * 需要执行super.process(item)才会调用自定义校验器
         */
        super.process(item);
        /**
         * 对数据进行简单的处理
         */
        if (item.getBlogItem().equals("springboot")) {
            item.setBlogTitle("springboot 系列还请看看我Jc");
        } else {
            item.setBlogTitle("未知系列");
        }
        return item;
    }
}
 
创建MyBeanValidator.java：
 
import org.springframework.batch.item.validator.ValidationException;
import org.springframework.batch.item.validator.Validator;
import org.springframework.beans.factory.InitializingBean;
import javax.validation.ConstraintViolation;
import javax.validation.Validation;
import javax.validation.ValidatorFactory;
import java.util.Set;
public class MyBeanValidator implements Validator, InitializingBean {
    private javax.validation.Validator validator;
    @Override
    public void validate(T value) throws ValidationException {
        /**
         * 使用Validator的validate方法校验数据
         */
        Set constraintViolations =
                validator.validate(value);
        if (constraintViolations.size() > 0) {
            StringBuilder message = new StringBuilder();
            for (ConstraintViolation constraintViolation : constraintViolations) {
                message.append(constraintViolation.getMessage() + "\n");
            }
            throw new ValidationException(message.toString());
        }
    }
    /**
     * 使用JSR-303的Validator来校验我们的数据，在此进行JSR-303的Validator的初始化
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        ValidatorFactory validatorFactory =
                Validation.buildDefaultValidatorFactory();
        validator = validatorFactory.usingContext().getValidator();
    }
}
 
ps：其实该篇文章没有使用这个数据校验器，大家想使用的话，可以在实体类上添加一些校验器的注解@NotNull @Max @Email等等。我偏向于直接在处理器里面进行处理，想把关于数据处理的代码都写在一块。
ItemWriter
写在MyBatchConfig类里
 
/**
 * ItemWriter定义：指定datasource，设置批量插入sql语句，写入数据库
 * @param dataSource
 * @return
 */
@Bean
public ItemWriter writer(DataSource dataSource){
    // 使用jdbcBcatchItemWrite写数据到数据库中
    JdbcBatchItemWriter writer = new JdbcBatchItemWriter();
    // 设置有参数的sql语句
    writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider());
    String sql = "insert into bloginfo "+" (blogAuthor,blogUrl,blogTitle,blogItem) "
            +" values(:blogAuthor,:blogUrl,:blogTitle,:blogItem)";
    writer.setSql(sql);
    writer.setDataSource(dataSource);
    return writer;
}
 
简单代码解析：
同样 对于数据读取器 ItemWriter ，我们给它也安排了一个输出监听器，创建 MyWriteListener.java：
 
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;
public class MyWriteListener implements ItemWriteListener {
    private Logger logger = LoggerFactory.getLogger(MyWriteListener.class);
    @Override
    public void beforeWrite(List