起源：blog.csdn.net/qq_35387940/article/details/108193473
前言
概念词就不多说了，我简略地介绍下 ， spring batch 是一个 方便使用的 较健全的 批处理 框架。
为什么说是方便使用的，因为这是 基于spring的一个框架，接入简略、易了解、流程明显。
为什么说是较健全的， 因为它提供了平常咱们在对大批量数据进行解决时须要思考到的 日志跟踪、事务粒度调配、可控执行、失败机制、重试机制、数据读写等。
注释
那么回到文章，咱们该篇文章将会带来给大家的是什么？（联合实例解说那是当然的）
从实现的业务场景来说，有以下两个：
- 从 csv文件 读取数据，进行业务解决再存储
- 从 数据库 读取数据，进行业务解决再存储
也就是平时常常遇到的数据清理或者数据过滤，又或者是数据迁徙备份等等。大批量的数据，本人实现分批解决须要思考的货色太多了，又不释怀，那么应用 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=5000server: port: 8665ps：这里咱们用到了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 : **/@Mapperpublic 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工作，蕴含一个或多个StepStep 蕴含（ItemReader、ItemProcessor和ItemWriter)ItemReader 数据读取器ItemProcessor 数据处理器ItemWriter 数据输入器首先，在MyBatchConfig类前退出注解：
@Configuration 用于通知spring，咱们这个类是一个自定义配置类，外面很多bean都须要加载到spring容器外面
@EnableBatchProcessing 开启批处理反对
而后开始往MyBatchConfig类里，编写各个小组件。
JobRepository
写在MyBatchConfig类里
/** * JobRepository定义：Job的注册容器以及和数据库打交道（事务管理等） * @param dataSource * @param transactionManager * @return * @throws Exception */@Beanpublic 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 */@Beanpublic 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 */@Beanpublic Job myJob(JobBuilderFactory jobs, Step myStep){ return jobs.get("myJob") .incrementer(new RunIdIncrementer()) .flow(myStep) .end() .listener(myJobListener()) .build();}对于Job的运行，是能够配置监听器的
JobListener
写在MyBatchConfig类里
/** * 注册job监听器 * @return */@Beanpublic 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 */@Beanpublic 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 */@Beanpublic ItemProcessor<BlogInfo, BlogInfo> processor(){ MyItemProcessor myItemProcessor = new MyItemProcessor(); // 设置校验器 myItemProcessor.setValidator(myBeanValidator()); return myItemProcessor;}数据处理器，是咱们自定义的，外面次要是蕴含咱们对数据处理的业务逻辑，并且咱们设置了一些数据校验器，咱们这里应用 JSR-303的Validator来作为校验器。
校验器
写在MyBatchConfig类里
/** * 注册校验器 * @return */@Beanpublic 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 */@Beanpublic 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 */@Beanpublic 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 : **/@RestControllerpublic 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工作，蕴含一个或多个StepStep 蕴含（ItemReader、ItemProcessor和ItemWriter)ItemReader 数据读取器ItemProcessor 数据处理器ItemWriter 数据输入器job 监听器reader 监听器writer 监听器process 数据校验器那么对于咱们新写一个job实现 一个新的场景，咱们须要全副重写么？
显然没必要，当然齐全新写一套也是能够的。
那么该篇，对于一个新的也出场景，从csv文件读取数据转换到数据库表读取数据，咱们从新新建的有：
- 数据读取器： 原先应用的是
FlatFileItemReader，咱们当初改为应用MyBatisCursorItemReader - 数据处理器： 新的场景，业务为了好扩大，所以咱们处理器最好也新建一个
- 数据输入器： 新的场景，业务为了好扩大，所以咱们数据输入器最好也新建一个
- step的绑定设置： 新的场景，业务为了好扩大，所以咱们step最好也新建一个
- 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 */@Beanpublic Job myJobNew(JobBuilderFactory jobs, Step stepNew){ return jobs.get("myJobNew") .incrementer(new RunIdIncrementer()) .flow(stepNew) .end() .listener(myJobListener()) .build();}@Beanpublic 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();}@Beanpublic ItemProcessor<BlogInfo, BlogInfo> processorNew(){ MyItemProcessorNew csvItemProcessor = new MyItemProcessorNew(); // 设置校验器 csvItemProcessor.setValidator(myBeanValidator()); return csvItemProcessor;}@Autowiredprivate 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 */@Beanpublic 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：
@AutowiredSimpleJobLauncher jobLauncher;@AutowiredJob 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开发手册（嵩山版）》最新公布，速速下载！
感觉不错，别忘了顺手点赞+转发哦！