关于java:Spring-Boot-Spring-Batch-实现批处理任务保姆级教程场景实战

48次阅读

共计 19313 个字符，预计需要花费 49 分钟才能阅读完成。

起源：blog.csdn.net/qq_35387940/article/details/108193473

前言

概念词就不多说了，我简略地介绍下，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 文件里的外围依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

<!--  spring batch -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-batch</artifactId>
</dependency>

<!-- hibernate validator -->
<dependency>
    <groupId>org.hibernate</groupId>
    <artifactId>hibernate-validator</artifactId>
    <version>6.0.7.Final</version>
</dependency>
<!--  mybatis -->
<dependency>
    <groupId>org.mybatis.spring.boot</groupId>
    <artifactId>mybatis-spring-boot-starter</artifactId>
    <version>2.0.0</version>
</dependency>
<!--  mysql -->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <scope>runtime</scope>
</dependency>

<!-- druid 数据源驱动 1.1.10 解决 springboot 从 1.0——2.0 版本问题 -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>druid-spring-boot-starter</artifactId>
    <version>1.1.18</version>
</dependency>

yml 文件：

Spring Boot 根底就不介绍了，举荐看这个实战我的项目：

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

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：

/**
 * @Author : JCccc
 * @Description :
 **/
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 + '\'' +
                '}';
    }

    public Integer getId() {return id;}

    public void setId(Integer id) {this.id = id;}

    public String getBlogAuthor() {return blogAuthor;}

    public void setBlogAuthor(String blogAuthor) {this.blogAuthor = blogAuthor;}

    public String getBlogUrl() {return blogUrl;}

    public void setBlogUrl(String blogUrl) {this.blogUrl = blogUrl;}

    public String getBlogTitle() {return blogTitle;}

    public void setBlogTitle(String blogTitle) {this.blogTitle = blogTitle;}

    public String getBlogItem() {return blogItem;}

    public void setBlogItem(String blogItem) {this.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;

/**
 * @Author : JCccc
 * @Description :
 **/
@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 < #{authorId}")
     List<BlogInfo> queryInfoById(Map<String , Integer> map);

}

接下来，重头戏，咱们开始对前边那张图里波及到的各个小组件进行编码。

首先创立一个 配置类，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<BlogInfo> reader(){
    // 应用 FlatFileItemReader 去读 cvs 文件，一行即一条数据
    FlatFileItemReader<BlogInfo> reader = new FlatFileItemReader<>();
    // 设置文件处在门路
    reader.setResource(new ClassPathResource("static/bloginfo.csv"));
    // entity 与 csv 数据做映射
    reader.setLineMapper(new DefaultLineMapper<BlogInfo>() {
        {setLineTokenizer(new DelimitedLineTokenizer() {
                {setNames(new String[]{"blogAuthor","blogUrl","blogTitle","blogItem"});
                }
            });
            setFieldSetMapper(new BeanWrapperFieldSetMapper<BlogInfo>() {
                {setTargetType(BlogInfo.class);
                }
            });
        }
    });
    return reader;
}

简略代码解析：

对于数据读取器 ItemReader，咱们给它安顿了一个读取监听器，创立 MyReadListener.java：

/**
 * @Author : JCccc
 * @Description :
 **/

public class MyReadListener implements ItemReadListener<BlogInfo> {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<BlogInfo, BlogInfo> processor(){MyItemProcessor myItemProcessor = new MyItemProcessor();
    // 设置校验器
    myItemProcessor.setValidator(myBeanValidator());
    return myItemProcessor;
}

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

校验器

写在 MyBatchConfig 类里 

/**
 * 注册校验器
 * @return
 */
@Bean
public MyBeanValidator myBeanValidator(){return new MyBeanValidator<BlogInfo>();
}

创立 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;

/**
 * @Author : JCccc
 * @Description :
 **/
public class MyItemProcessor extends ValidatingItemProcessor<BlogInfo> {
    @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;

/**
 * @Author : JCccc
 * @Description :
 **/
public class MyBeanValidator<T> implements Validator<T>, InitializingBean {

    private javax.validation.Validator validator;

    @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());
        }
    }

    /**
     * 应用 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<BlogInfo> writer(DataSource dataSource){
    // 应用 jdbcBcatchItemWrite 写数据到数据库中
    JdbcBatchItemWriter<BlogInfo> writer = new JdbcBatchItemWriter<>();
    // 设置有参数的 sql 语句
    writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<BlogInfo>());
    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;

/**
 * @Author : JCccc
 * @Description :
 **/
public class MyWriteListener implements ItemWriteListener<BlogInfo> {private Logger logger = LoggerFactory.getLogger(MyWriteListener.class);

    @Override
    public void beforeWrite(List<? extends BlogInfo> items) { }

    @Override
    public void afterWrite(List<? extends BlogInfo> items) { }

    @Override
    public void onWriteError(Exception exception, List<? extends BlogInfo> items) {
        try {logger.info(format("%s%n", exception.getMessage()));
            for (BlogInfo message : items) {logger.info(format("Failed writing BlogInfo : %s", message.toString()));
            }

        } catch (Exception e) {e.printStackTrace();
        }

    }
}

ItemReader、ItemProcessor、ItemWriter，这三个小组件到这里，咱们都实现了，那么接下来就是把这三个小组件跟咱们的 step 去绑定起来。

写在 MyBatchConfig 类里 

/**
 * step 定义：* 包含
 * ItemReader 读取
 * ItemProcessor  解决
 * ItemWriter 输入
 * @param stepBuilderFactory
 * @param reader
 * @param writer
 * @param processor
 * @return
 */

@Bean
public Step myStep(StepBuilderFactory stepBuilderFactory, ItemReader<BlogInfo> reader,
                 ItemWriter<BlogInfo> writer, ItemProcessor<BlogInfo, BlogInfo> processor){
    return stepBuilderFactory
            .get("myStep")
            .<BlogInfo, BlogInfo>chunk(65000) // Chunk 的机制 (即每次读取一条数据，再解决一条数据，累积到肯定数量后再一次性交给 writer 进行写入操作)
            .reader(reader).faultTolerant().retryLimit(3).retry(Exception.class).skip(Exception.class).skipLimit(2)
            .listener(new MyReadListener())
            .processor(processor)
            .writer(writer).faultTolerant().skip(Exception.class).skipLimit(2)
            .listener(new MyWriteListener())
            .build();}

这个 Step，稍作解说。

前边提到了，spring batch 框架，提供了事务的管制，重启，检测跳过等等机制。

那么，这些货色的实现，很多都在于这个 step 环节的设置。

首先看到咱们代码呈现的第一个设置，chunk(6500) ，Chunk 的机制 (即每次读取一条数据，再解决一条数据，累积到肯定数量后再一次性交给 writer 进行写入操作。

没错，对于整个 step 环节，就是数据的读取，解决最初到输入。

这个 chunk 机制里，咱们传入的 6500，也就是是通知它，读取解决数据，累计达到 6500 条进行一次批次解决，去执行写入操作。

这个传值，是依据具体业务而定，能够是 500 条一次，1000 条一次，也能够是 20 条一次，50 条一次。

通过一张简略的小图来帮忙了解：

在咱们大量数据处理，不论是读取或者说是写入，都必定会波及到一些未知或者已知因素导致某条数据失败了。

那么如果说咱们啥也不设置，失败一条数据，那么咱们就当作整个失败了？。显然这个太不兽性，所以 spring batch 提供了 retry 和 skip 两个设置（其实还有 restart），通过这两个设置来人性化地解决一些数据操作失败场景。

retryLimit(3).retry(Exception.class)

没错，这个就是设置重试，当出现异常的时候，重试多少次。咱们设置为 3，也就是说当一条数据操作失败，那咱们会对这条数据进行重试 3 次，还是失败就是 当做失败了，那么咱们如果有配置 skip（举荐配置应用），那么这个数据失败记录就会留到给 skip 来解决。

skip(Exception.class).skipLimit(2)

skip，跳过，也就是说咱们如果设置 3，那么就是能够容忍 3 条数据的失败。只有达到失败数据达到 3 次，咱们才中断这个 step。

对于失败的数据，咱们做了相干的监听器以及异样信息记录，供与后续手动补救。

那么记下来咱们开始去调用这个批处理 job，咱们通过接口去触发这个批处理事件，新建一个 Controller，TestController.java：

/**
 * @Author : JCccc
 * @Description :
 **/
@RestController
public class TestController {
    @Autowired
    SimpleJobLauncher jobLauncher;

    @Autowired
    Job myJob;

    @GetMapping("testJob")
    public  void testJob() throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
     //    后置参数：应用 JobParameters 中绑定参数 addLong  addString 等办法
        JobParameters jobParameters = new JobParametersBuilder().toJobParameters();
        jobLauncher.run(myJob, jobParameters);

    }
}

对了，我筹备了一个 csv 文件 bloginfo.csv，外面大略 8 万多条数据，用来进行批处理测试：

这个文件的门路跟咱们的数据读取器外面读取的门路要始终，

目前咱们数据库是这个样子，

接下来咱们把咱们的我的项目启动起来，再看一眼数据库，生成了一些 batch 用来跟踪记录 job 的一些数据表：

咱们来调用一下 testJob 接口，

而后看下数据库，能够看的数据全副都进行了相干的逻辑解决并插入到了数据库：

到这里，咱们对 Springboot 整合 spring batch 其实曾经操作结束了，也实现了从 csv 文件读取数据解决存储的业务场景。

从数据库读取数据

ps：前排提醒应用 druid 有坑。前面会讲到。

那么接下来实现场景，从数据库表内读取数据进行解决输入到新的表外面。

那么基于咱们上边的整合，咱们曾经实现了 

JobRepository job 的注册 / 存储器
JobLauncher job 的执行器
Job job 工作，蕴含一个或多个 Step
Step 蕴含（ItemReader、ItemProcessor 和 ItemWriter)
ItemReader 数据读取器
ItemProcessor 数据处理器
ItemWriter 数据输入器
job 监听器
reader 监听器
writer 监听器
process 数据校验器

那么对于咱们新写一个 job 实现 一个新的场景，咱们须要全副重写么？

显然没必要，当然齐全新写一套也是能够的。

那么该篇，对于一个新的也出场景，从 csv 文件读取数据转换到数据库表读取数据，咱们从新新建的有：

  1. 数据读取器： 原先应用的是 FlatFileItemReader，咱们当初改为应用 MyBatisCursorItemReader
  2. 数据处理器： 新的场景，业务为了好扩大，所以咱们处理器最好也新建一个
  3. 数据输入器： 新的场景，业务为了好扩大，所以咱们数据输入器最好也新建一个
  4. step 的绑定设置： 新的场景，业务为了好扩大，所以咱们 step 最好也新建一个
  5. Job： 当然是要从新写一个了

其余咱们照用原先的就行，JobRepository，JobLauncher 以及各种监听器啥的，暂且不从新建了。

新建 MyItemProcessorNew.java：

import org.springframework.batch.item.validator.ValidatingItemProcessor;
import org.springframework.batch.item.validator.ValidationException;

/**
 * @Author : JCccc
 * @Description :
 **/
public class MyItemProcessorNew extends ValidatingItemProcessor<BlogInfo> {
    @Override
    public BlogInfo process(BlogInfo item) throws ValidationException {
        /**
         * 须要执行 super.process(item) 才会调用自定义校验器
         */
        super.process(item);
        /**
         * 对数据进行简略的解决
         */
        Integer authorId= Integer.valueOf(item.getBlogAuthor());
        if (authorId<20000) {item.setBlogTitle("这是都是小于 20000 的数据");
        } else if (authorId>20000 && authorId<30000){item.setBlogTitle("这是都是小于 30000 然而大于 20000 的数据");
        }else {item.setBlogTitle("新书不厌百回读");
        }
        return item;
    }
}

而后其余从新定义的小组件，写在 MyBatchConfig 类里：

/**
 * 定义 job
 * @param jobs
 * @param stepNew
 * @return
 */
@Bean
public Job myJobNew(JobBuilderFactory jobs, Step stepNew){return jobs.get("myJobNew")
            .incrementer(new RunIdIncrementer())
            .flow(stepNew)
            .end()
            .listener(myJobListener())
            .build();}

@Bean
public Step stepNew(StepBuilderFactory stepBuilderFactory, MyBatisCursorItemReader<BlogInfo> itemReaderNew,
                    ItemWriter<BlogInfo> writerNew, ItemProcessor<BlogInfo, BlogInfo> processorNew){
    return stepBuilderFactory
            .get("stepNew")
            .<BlogInfo, BlogInfo>chunk(65000) // Chunk 的机制 (即每次读取一条数据，再解决一条数据，累积到肯定数量后再一次性交给 writer 进行写入操作)
            .reader(itemReaderNew).faultTolerant().retryLimit(3).retry(Exception.class).skip(Exception.class).skipLimit(10)
            .listener(new MyReadListener())
            .processor(processorNew)
            .writer(writerNew).faultTolerant().skip(Exception.class).skipLimit(2)
            .listener(new MyWriteListener())
            .build();}

@Bean
public ItemProcessor<BlogInfo, BlogInfo> processorNew(){MyItemProcessorNew csvItemProcessor = new MyItemProcessorNew();
    // 设置校验器
    csvItemProcessor.setValidator(myBeanValidator());
    return csvItemProcessor;
}

@Autowired
private SqlSessionFactory sqlSessionFactory;

@Bean
@StepScope
//Spring Batch 提供了一个非凡的 bean scope 类（StepScope: 作为一个自定义的 Spring bean scope）。这个 step scope 的作用是连贯 batches 的各个 steps。这个机制容许配置在 Spring 的 beans 当 steps 开始时才实例化并且容许你为这个 step 指定配置和参数。public MyBatisCursorItemReader<BlogInfo> itemReaderNew(@Value("#{jobParameters[authorId]}") String authorId) {System.out.println("开始查询数据库");

        MyBatisCursorItemReader<BlogInfo> reader = new MyBatisCursorItemReader<>();

        reader.setQueryId("com.example.batchdemo.mapper.BlogMapper.queryInfoById");

        reader.setSqlSessionFactory(sqlSessionFactory);
         Map<String , Object> map = new HashMap<>();

          map.put("authorId" , Integer.valueOf(authorId));
         reader.setParameterValues(map);
        return reader;
}

/**
 * ItemWriter 定义：指定 datasource，设置批量插入 sql 语句，写入数据库
 * @param dataSource
 * @return
 */
@Bean
public ItemWriter<BlogInfo> writerNew(DataSource dataSource){
    // 应用 jdbcBcatchItemWrite 写数据到数据库中
    JdbcBatchItemWriter<BlogInfo> writer = new JdbcBatchItemWriter<>();
    // 设置有参数的 sql 语句
    writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<BlogInfo>());
    String sql = "insert into bloginfonew"+"(blogAuthor,blogUrl,blogTitle,blogItem)"
            +"values(:blogAuthor,:blogUrl,:blogTitle,:blogItem)";
    writer.setSql(sql);
    writer.setDataSource(dataSource);
    return writer;
}

代码须要留神的点

数据读取器 MyBatisCursorItemReader

对应的 mapper 办法：

数据处理器 MyItemProcessorNew：

数据输入器，新插入到别的数据库表去，特意这样为了测试：

当然咱们的数据库为了测试这个场景，也是新建了一张表，bloginfonew 表。

接下来，咱们新写一个接口来执行新的这个 job：

@Autowired
SimpleJobLauncher jobLauncher;

@Autowired
Job myJobNew;

@GetMapping("testJobNew")
public  void testJobNew(@RequestParam("authorId") String authorId) throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {JobParameters jobParametersNew = new JobParametersBuilder().addLong("timeNew", System.currentTimeMillis())
            .addString("authorId",authorId)
            .toJobParameters();
    jobLauncher.run(myJobNew,jobParametersNew);

}

ok，咱们来调用一些这个接口：

看下控制台：

没错，这就是失败的，起因是因为跟 druid 无关，报了一个数据库性能不反对。这是在数据读取的时候报的错。

我初步测试认为是 MyBatisCursorItemReader，druid 数据库连接池不反对。

那么，咱们只须要：

正文掉 druid 连接池 jar 依赖

yml 里替换连接池配置

其实咱们不配置其余连接池，springboot 2.X 版本曾经为咱们整合了默认的连接池 HikariCP。

在 Springboot2.X 版本，数据库的连接池官网举荐应用 HikariCP

如果不是为了 druid 的那些后盾监控数据，sql 剖析等等，齐全是优先应用 HikariCP 的。

官网的原话：

We preferHikariCPfor its performance and concurrency. If HikariCP is available, we always choose it.

翻译：

咱们更喜爱 hikaricpf 的性能和并发性。如果有 HikariCP，咱们总是抉择它。

所以咱们就啥连接池也不配了，应用默认的 HikariCP 连接池。

举荐一个开源收费的 Spring Boot 实战我的项目：

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

当然你想配，也是能够的：

所以咱们剔除掉 druid 链接池后，咱们再来调用一下新接口：

能够看到，从数据库获取数据并进行批次解决写入 job 是胜利的：

新的表外面插入的数据都进行了本人写的逻辑解决：

好了，springboot 整合 spring batch 批处理框架，就到此吧。

近期热文举荐：

1.1,000+ 道 Java 面试题及答案整顿 (2022 最新版)

2. 劲爆！Java 协程要来了。。。

3.Spring Boot 2.x 教程，太全了！

4. 别再写满屏的爆爆爆炸类了，试试装璜器模式，这才是优雅的形式！！

5.《Java 开发手册（嵩山版）》最新公布，速速下载！

感觉不错，别忘了顺手点赞 + 转发哦！

正文完
java
发表至： java
2023-08-24
 0
一个-static-还能难得住我
跃迁之路850天程序员高效学习方法论探索系列实验阶段607201972024
关于java:爱了Java知识点最强总结美女学姐带你入坑网友感叹真香
关于java:Java单元测试技巧之PowerMock