Spring Boot + Spring Batch 实现批处理任务，保姆级教程！（场景实战）
来源：blog.csdn.net/qq_35387940/article/details/108193473前言
概念词就不多说了，我简单地介绍下 ， spring batch 是一个 方便使用的 较健全的 批处理 框架。
为什么说是方便使用的，因为这是 基于spring的一个框架，接入简单、易理解、流程分明。
为什么说是较健全的， 因为它提供了往常我们在对大批量数据进行处理时需要考虑到的 日志跟踪、事务粒度调配、可控执行、失败机制、重试机制、数据读写等。
正文
那么回到文章，我们该篇文章将会带来给大家的是什么？（结合实例讲解那是当然的）
从实现的业务场景来说，有以下两个：
[*]从csv文件 读取数据，进行业务处理再存储
[*]从 数据库 读取数据，进行业务处理再存储
也就是平时经常遇到的数据清理或者数据过滤，又或者是数据迁移备份等等。大批量的数据，自己实现分批处理需要考虑的东西太多了，又不放心，那么使用 Spring Batch 框架 是一个很好的选择。
首先，在进入实例教程前，我们看看这次的实例里，我们使用springboot 整合spring batch 框架，要编码的东西有什么？
通过一张简单的图来了解：
https://img2023.cnblogs.com/other/1218593/202308/1218593-20230824090503286-1734553250.png
可能大家看到这个图，是不是多多少少想起来定时任务框架？确实有那么点像，但是我必须在这告诉大家，这是一个批处理框架，不是一个schuedling 框架。但是前面提到它提供了可执行控制，也就是说，啥时候执行是可控的，那么显然就是自己可以进行扩展结合定时任务框架，实现你心中所想。
ok，回到主题，相信大家能从图中简单明了地看到我们这次实例，需要实现的东西有什么了。所以我就不在对各个小组件进行大批量文字的描述了。
那么我们事不宜迟，开始我们的实例教程。
首先准备一个数据库，里面建一张简单的表，用于实例数据的写入存储或者说是读取等等。
bloginfo表
https://img2023.cnblogs.com/other/1218593/202308/1218593-20230824090503638-1099570515.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>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<dependency>
<groupId>org.hibernate</groupId>
<artifactId>hibernate-validator</artifactId>
<version>6.0.7.Final</version>
</dependency>
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>2.0.0</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<scope>runtime</scope>
</dependency>
<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: 8665https://img2023.cnblogs.com/other/1218593/202308/1218593-20230824090503985-702402807.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;
@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 开启批处理支持
https://img2023.cnblogs.com/other/1218593/202308/1218593-20230824090504392-1648689200.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();
// 设置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（ItemReaderItemProcessorItemWriter）
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;
}简单代码解析：
https://img2023.cnblogs.com/other/1218593/202308/1218593-20230824090505051-706403639.png
对于数据读取器 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;
}简单代码解析：
https://img2023.cnblogs.com/other/1218593/202308/1218593-20230824090505714-643938961.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 { private Logger logger = LoggerFactory.getLogger(MyWriteListener.class); @Override public void beforeWrite(List
页:
[1]