批处理框架 Spring Batch 这么强，你真的会用吗？

2023-11-10 14:51
文章标签 java spring 框架 真的 批处理 batch 会用

本文主要是介绍批处理框架 Spring Batch 这么强，你真的会用吗？，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

点击上方“芋道源码”，选择“设为星标”

管她前浪，还是后浪？

能浪的浪，才是好浪！

每天 10:33 更新文章，每天掉亿点点头发...

源码精品专栏

 

  • 原创 | Java 2021 超神之路，很肝~

  • 中文详细注释的开源项目

  • RPC 框架 Dubbo 源码解析

  • 网络应用框架 Netty 源码解析

  • 消息中间件 RocketMQ 源码解析

  • 数据库中间件 Sharding-JDBC 和 MyCAT 源码解析

  • 作业调度中间件 Elastic-Job 源码解析

  • 分布式事务中间件 TCC-Transaction 源码解析

  • Eureka 和 Hystrix 源码解析

  • Java 并发源码

来源：blog.csdn.net/qq_35387940/

article/details/108193473

  • 前言

  • 正文

前言

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

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

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

基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能

  • 项目地址：https://github.com/YunaiV/ruoyi-vue-pro

  • 视频教程：https://doc.iocoder.cn/video/

正文

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

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

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

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

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

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

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

fcfaa4d49e342fd019dc35ea9fe0c317.png

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

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

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

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

bloginfo表

90a87e1e41d83600d18ddf3960585439.png

相关建表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:batch:job:
#设置为 false -需要jobLaucher.run执行enabled: falseinitialize-schema: always> 基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
>
> * 项目地址：<https://github.com/YunaiV/yudao-cloud>
> * 视频教程：<https://doc.iocoder.cn/video/>#    table-prefix: my-batchdatasource:druid:username: rootpassword: rooturl: jdbc:mysql://localhost:3306/hellodemo?useSSL=false&useUnicode=true&characterEncoding=UTF-8&serverTimezone=GMT%2B8&zeroDateTimeBehavior=convertToNulldriver-class-name: com.mysql.cj.jdbc.DriverinitialSize: 5minIdle: 5maxActive: 20maxWait: 60000timeBetweenEvictionRunsMillis: 60000minEvictableIdleTimeMillis: 300000validationQuery: SELECT 1 FROM DUALtestWhileIdle: truetestOnBorrow: falsetestOnReturn: falsepoolPreparedStatements: truemaxPoolPreparedStatementPerConnectionSize: 20useGlobalDataSourceStat: trueconnectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000
server:port: 8665
a8206cb3e08bc2e5bdc53b6e2a7c5c9c.png

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;@Overridepublic 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 开启批处理支持

d36c42cc501824c8a9af8e5a0f43d646.png

然后开始往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();// 设置jobRepositoryjobLauncher.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);@Overridepublic void beforeJob(JobExecution jobExecution) {logger.info("job 开始, id={}",jobExecution.getJobId());}@Overridepublic 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;
}

简单代码解析：

4a697289b51ea3e72d8100a54485397f.png

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

/*** @Author : JCccc* @Description :**/public class MyReadListener implements ItemReadListener<BlogInfo> {private Logger logger = LoggerFactory.getLogger(MyReadListener.class);@Overridepublic void beforeRead() {}@Overridepublic void afterRead(BlogInfo item) {}@Overridepublic 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> {@Overridepublic 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;@Overridepublic 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*/@Overridepublic 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;
}

简单代码解析：

bc4d93727b79b7ff1f4f76ffa2220dcb.png

同样 对于数据读取器 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);@Overridepublic void beforeWrite(List<? extends BlogInfo> items) {}@Overridepublic void afterWrite(List<? extends BlogInfo> items) {}@Overridepublic 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条一次。

通过一张简单的小图来帮助理解：

166df161c2c6af278973848cc3decb78.png

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

那么如果说咱们啥也不设置，失败一条数据，那么我们就当作整个失败了？。显然这个太不人性，所以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 {@AutowiredSimpleJobLauncher jobLauncher;@AutowiredJob 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万多条数据，用来进行批处理测试：

0e5780b1464ce2f437bb0d3476df1bd0.png

这个文件的路径跟我们的数据读取器里面读取的路径要一直，

360ed3d8792962f5d7dfe4d4f9072d0e.png

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

a7b0fff2d19a6cea9621016bde5f560e.png

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

e93723ac89792c606d2b56489b80b2ec.png

我们来调用一下testJob接口，

f6936beda67049f69596cef9f6659dfe.png

然后看下数据库，可以看的数据全部都进行了相关的逻辑处理并插入到了数据库：

eb6b4dcbe7e1844e1a8504cb8e9641c9.png

到这里，我们对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> {@Overridepublic 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

7af16ea1cce4fb878c372dd3c6c5661a.png

对应的mapper方法：

b5287788606b517f0a8ed6ed315f435b.png

数据处理器 MyItemProcessorNew：

ee1277bdaa837eb8f2cbe376fce65ac5.png

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

992ca06d332637dba86f9c0d8b04d780.png

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

af9788a7a27f11b9600155b0c63017c3.png

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

e5c08d31102d36009ed3501b172f681d.png 
@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，我们来调用一些这个接口：

1866562029bed0dfdbb7cb3656d8ab1c.png

看下控制台：

51f49808a75bde412a82b1e7e191ce69.png

没错，这就是失败的，原因是因为跟druid有关，报了一个数据库功能不支持。这是在数据读取的时候报的错。

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

那么，我们只需要：

  1. 注释掉druid连接池 jar依赖

4922873f9029dd6b6ea86e2674317fae.png

  1. yml里替换连接池配置

9991e57bb244efffbdf94cee24be72e0.png

其实我们不配置其他连接池，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 连接池。

当然你想配，也是可以的：

c2484cb75468f4239e4b3bc408a3e9d6.png

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

9d6593f2a00a481645a1dbc82831aa39.png

可以看到，从数据库获取数据并进行批次处理写入job是成功的：

0d905394880a37c58b31c625d8c37c48.png

新的表里面插入的数据都进行了自己写的逻辑处理：

f1ae55e64a945af6a27c9ea746ec063f.png

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

欢迎加入我的知识星球，一起探讨架构，交流源码。加入方式，长按下方二维码噢：

f6993c834733f6ccb28653002f7f7a81.png

已在知识星球更新源码解析如下：

d702dcffe1971c1da2b4be66ff5cf636.jpeg

16f560cb84c9adf5c1119928d917ae4a.jpeg

05f9552eaaef4f1e4b173346638880f2.jpeg

9595b256a285a192239f8b930c87783d.jpeg

最近更新《芋道 SpringBoot 2.X 入门》系列，已经 101 余篇，覆盖了 MyBatis、Redis、MongoDB、ES、分库分表、读写分离、SpringMVC、Webflux、权限、WebSocket、Dubbo、RabbitMQ、RocketMQ、Kafka、性能测试等等内容。

提供近 3W 行代码的 SpringBoot 示例，以及超 4W 行代码的电商微服务项目。

获取方式：点“在看”，关注公众号并回复 666 领取，更多内容陆续奉上。

文章有帮助的话，在看，转发吧。
谢谢支持哟 (*^__^*）

这篇关于批处理框架 Spring Batch 这么强，你真的会用吗？的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/383396。

相关文章

springboot健康检查监控全过程

springboot健康检查监控全过程

《springboot健康检查监控全过程》文章介绍了SpringBoot如何使用Actuator和Micrometer进行健康检查和监控,通过配置和自定义健康指示器,开发者可以实时监控应用组件的状态,... 目录1. 引言重要性2. 配置Spring Boot ActuatorSpring Boot Act
阅读更多...
使用Java解析JSON数据并提取特定字段的实现步骤(以提取mailNo为例)

使用Java解析JSON数据并提取特定字段的实现步骤(以提取mailNo为例)

《使用Java解析JSON数据并提取特定字段的实现步骤(以提取mailNo为例)》在现代软件开发中,处理JSON数据是一项非常常见的任务,无论是从API接口获取数据,还是将数据存储为JSON格式,解析... 目录1. 背景介绍1.1 jsON简介1.2 实际案例2. 准备工作2.1 环境搭建2.1.1 添加
阅读更多...
Java实现任务管理器性能网络监控数据的方法详解

Java实现任务管理器性能网络监控数据的方法详解

《Java实现任务管理器性能网络监控数据的方法详解》在现代操作系统中,任务管理器是一个非常重要的工具,用于监控和管理计算机的运行状态,包括CPU使用率、内存占用等,对于开发者和系统管理员来说,了解这些... 目录引言一、背景知识二、准备工作1. Maven依赖2. Gradle依赖三、代码实现四、代码详解五
阅读更多...
java如何分布式锁实现和选型

java如何分布式锁实现和选型

《java如何分布式锁实现和选型》文章介绍了分布式锁的重要性以及在分布式系统中常见的问题和需求,它详细阐述了如何使用分布式锁来确保数据的一致性和系统的高可用性,文章还提供了基于数据库、Redis和Zo... 目录引言：分布式锁的重要性与分布式系统中的常见问题和需求分布式锁的重要性分布式系统中常见的问题和需求
阅读更多...
SpringBoot基于MyBatis-Plus实现Lambda Query查询的示例代码

SpringBoot基于MyBatis-Plus实现Lambda Query查询的示例代码

《SpringBoot基于MyBatis-Plus实现LambdaQuery查询的示例代码》MyBatis-Plus是MyBatis的增强工具,简化了数据库操作,并提高了开发效率,它提供了多种查询方... 目录引言基础环境配置依赖配置（Maven）application.yml 配置表结构设计demo_st
阅读更多...
在Ubuntu上部署SpringBoot应用的操作步骤

在Ubuntu上部署SpringBoot应用的操作步骤

《在Ubuntu上部署SpringBoot应用的操作步骤》随着云计算和容器化技术的普及,Linux服务器已成为部署Web应用程序的主流平台之一,Java作为一种跨平台的编程语言,具有广泛的应用场景,本... 目录一、部署准备二、安装 Java 环境1. 安装 JDK2. 验证 Java 安装三、安装 mys
阅读更多...
Springboot的ThreadPoolTaskScheduler线程池轻松搞定15分钟不操作自动取消订单

Springboot的ThreadPoolTaskScheduler线程池轻松搞定15分钟不操作自动取消订单

《Springboot的ThreadPoolTaskScheduler线程池轻松搞定15分钟不操作自动取消订单》：本文主要介绍Springboot的ThreadPoolTaskScheduler线... 目录ThreadPoolTaskScheduler线程池实现15分钟不操作自动取消订单概要1，创建订单后
阅读更多...
JAVA中整型数组、字符串数组、整型数和字符串 的创建与转换的方法

JAVA中整型数组、字符串数组、整型数和字符串 的创建与转换的方法

《JAVA中整型数组、字符串数组、整型数和字符串的创建与转换的方法》本文介绍了Java中字符串、字符数组和整型数组的创建方法,以及它们之间的转换方法,还详细讲解了字符串中的一些常用方法,如index... 目录一、字符串、字符数组和整型数组的创建1、字符串的创建方法1.1 通过引用字符数组来创建字符串1.2
阅读更多...
SpringCloud集成AlloyDB的示例代码

SpringCloud集成AlloyDB的示例代码

《SpringCloud集成AlloyDB的示例代码》AlloyDB是GoogleCloud提供的一种高度可扩展、强性能的关系型数据库服务,它兼容PostgreSQL,并提供了更快的查询性能... 目录1.AlloyDBjavascript是什么？AlloyDB 的工作原理2.搭建测试环境3.代码工程1.
阅读更多...
Java调用Python代码的几种方法小结

Java调用Python代码的几种方法小结

《Java调用Python代码的几种方法小结》Python语言有丰富的系统管理、数据处理、统计类软件包,因此从java应用中调用Python代码的需求很常见、实用,本文介绍几种方法从java调用Pyt... 目录引言Java core使用ProcessBuilder使用Java脚本引擎总结引言python
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2