乐趣区

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

起源: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();
        }

    }
}

ItemReaderItemProcessorItemWriter,这三个小组件到这里,咱们都实现了,那么接下来就是把这三个小组件跟咱们的 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 开发手册(嵩山版)》最新公布,速速下载!

感觉不错,别忘了顺手点赞 + 转发哦!

退出移动版