SpringBatch 自定义ItemReader和可重新启动Reader(十五)

文章目录

一、自定义CustomItemReader
二、job 监听器
三、配置job
四、改造CustomItemReader，发生异常批处理作业从停止的地方重新启动

前言：在一些业务场景中，可能现有的reader不符合我们的要求，SpringBatch提供自定义reader，实现ItemReader接口，满足我们业务场景。

SpringBatch其它文章直通车：

SpringBatch读单个文件(FlatFileItemReader)和写单个文件(FlatFileItemWriter)(一)
SpringBatch顺序读取多文件(MultiResourceItemReader)和顺序写文件(MultiResourceItemWriter)(二)
SpringBatch读数据库(MyBatisPagingItemReader)(三)
SpringBatch读文件(FlatFileItemReader)写据库(MyBatisBatchItemWriter)(四)
SpringBatch 监听器之Job监听器(JobExecutionListener)和Step监听器(StepExecutionListener)(五)
SpringBatch 监听器之Chunk监听器(ChunkListener)和Skip监听器(SkipListener)(六)
SpringBatch 多线程(TaskExecutor)启动Job详解 (七)
SpringBatch 配置并行启动Job详解 (八)
SpringBatch 批处理分区(Partitioner )分片(九)
SpringBatch tasklet实现和用法(十)
SpringBatch 读取JSON(JsonItemReader)用法(十一)
SpringBatch 写文件JSON(JsonFileItemWriter)用法(十二)
SpringBatch 读取xml文件(StaxEventItemReader)用法(十三)
SpringBatch 写xml文件(StaxEventItemWriter)用法(十四)

代码已上传GitHub上面地址：git源码地址

一、自定义CustomItemReader

创建了一个简单的ItemReader实现，它从提供的列表中读取数据。我们首先实现ItemReader最基本的read方法

package com.sl.common;import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.NonTransientResourceException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;import java.util.List;/*** 自定义reader* @author shuliangzhao* @Title: CustomItemReader* @ProjectName spring-boot-learn* @Description: TODO* @date 2019/9/21 14:49*/
public class CustomItemReader<T> implements ItemReader<T> {private List<String> list;public CustomItemReader(List<String> list) {this.list = list;}@Overridepublic T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {if (list.size() > 0 && !list.isEmpty()) {return (T)list.remove(0);}return null;}
}

二、job 监听器

job执行之前加载数据供reader使用

package com.sl.listener;import com.sl.common.CommonConstants;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.stereotype.Component;/*** @author shuliangzhao* @Title: CustomJobListener* @ProjectName spring-boot-learn* @Description: TODO* @date 2019/9/21 14:59*/
@Component
public class CustomJobListener implements JobExecutionListener {@Overridepublic void beforeJob(JobExecution jobExecution) {CommonConstants.getList().add("hello");CommonConstants.getList().add("springbatch");}@Overridepublic void afterJob(JobExecution jobExecution) {}
}

三、配置job

package com.sl.config;import com.sl.common.CommonConstants;
import com.sl.common.CustomItemReader;
import com.sl.listener.CustomJobListener;
import com.sl.writer.CustomItemWriter;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 自定义重启reader* @author shuliangzhao* @Title: CustomConfiguration* @ProjectName spring-boot-learn* @Description: TODO* @date 2019/9/21 14:55*/
@Configuration
@EnableBatchProcessing
public class CustomConfiguration {@Autowiredprivate JobBuilderFactory jobBuilderFactory;@Autowiredprivate StepBuilderFactory stepBuilderFactory;@Autowiredprivate CustomJobListener customJobListener;@Autowiredprivate CustomItemWriter customItemWriter;@Beanpublic Job customJob() {return jobBuilderFactory.get("customJob").listener(customJobListener).start(customStep()).build();}@Beanpublic Step customStep() {return stepBuilderFactory.get("customStep").chunk(10).reader(customItemReader()).writer(customItemWriter).build();}@Bean@StepScopepublic CustomItemReader customItemReader() {return new CustomItemReader(CommonConstants.getList());}
}

敲黑板：如果这个job在执行过程中发生错误，是不能重启。接下来我们讲解怎么改进reader可以让我们的job在发生异常，批处理作业从停止的地方重新启动。

四、改造CustomItemReader，发生异常批处理作业从停止的地方重新启动

目前，如果处理被中断并重新开始，ItemReader必须从头开始。在许多场景中，这实际上是有效的，但有时更可取的做法是，批处理作业从停止的地方重新启动。关键的区别通常是阅读器是有状态的还是无状态的。无状态读取器不需要担心可重启性，但是有状态读取器必须尝试在重启时重新构建其最后一个已知状态。出于这个原因，我们建议尽可能保持自定义读取器处于无状态，因此不必担心可重启性。

CustomItemReader还需要实现接口ItemStream

package com.sl.common;import org.springframework.batch.item.*;import java.util.List;/*** 自定义reader* @author shuliangzhao* @Title: CustomItemReader* @ProjectName spring-boot-learn* @Description: TODO* @date 2019/9/21 14:49*/
public class CustomRestaItemReader<T> implements ItemReader<T> , ItemStream {private List<T> list;int currentIndex = 0;private static final String CURRENT_INDEX = "current.index";public CustomRestaItemReader(List<T> list) {this.list = list;}@Overridepublic T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {if (currentIndex < list.size()) {return list.get(currentIndex++);}return null;}@Overridepublic void open(ExecutionContext executionContext) throws ItemStreamException {if(executionContext.containsKey(CURRENT_INDEX)){currentIndex = new Long(executionContext.getLong(CURRENT_INDEX)).intValue();}else{currentIndex = 0;}}@Overridepublic void update(ExecutionContext executionContext) throws ItemStreamException {executionContext.putLong(CURRENT_INDEX, new Long(currentIndex).longValue());}@Overridepublic void close() throws ItemStreamException {}
}

实现reader重启关键方法是open方法和update方法

SpringBatch 自定义ItemReader和可重新启动Reader(十五)相关推荐

FreeSql （三十五）CodeFirst 自定义特性
比如项目内已经使用了其它 orm,如 efcore,这样意味着实体中可能存在 [Key],但它与 FreeSql [Column(IsPrimary = true] 不同. Q: FreeSql 实体 ...
十五、Typora官方主题 + 自定义主题
十五.Typora官方主题 + 自定义主题 1.下载官方主题 2.在官方主题的基础上自定义 1.下载官方主题首先去Typora官网下载自己喜欢的主题:官方主题下载例如我这里选择 Vue 我们只用到 ...
[系统安全] 四十五.APT系列(10)Metasploit后渗透技术信息收集、权限提权和功能模块详解
您可能之前看到过我写的类似文章,为什么还要重复撰写呢?只是想更好地帮助初学者了解病毒逆向分析和系统安全,更加成体系且不破坏之前的系列.因此,我重新开设了这个专栏,准备系统整理和深入学习系统安全.逆向分 ...
十五步骤手把手学会制作网吧XP母盘制作
中国网吧在线 -> 网吧大学 - 技术学院 - 网管技术 - 正文十五步骤手把手学会制作网吧XP母盘制作 2008-4-9 12:35:05 来源: 中国网吧在线编辑:陈峰 [网友评论] ...
WCF技术剖析之十五：数据契约代理（DataContractSurrogate）在序列化中的作用
如果一个类型,不一定是数据契约,和给定的数据契约具有很大的差异,而我们要将该类型的对象序列化成基于数据契约对应的XML.反之,对于一段给定的基于数据契约的XML,要通过反序列化生成该类型的对象,我们该 ...
第十五章 IO流(转换流字符流字符缓冲流打印流)
Java基础15 第十五章 IO流(转换流字符流字符缓冲流打印流) 15.1 字符编码和字符集 15.1.1 字符编码 15.1.2 字符集 15.1.3 String类getBytes()方法 ...
第十五章使用管理门户SQL接口（一）
文章目录第十五章使用管理门户SQL接口(一) 管理门户SQL工具选择命名空间用户自定义执行SQL查询编写SQL语句表拖放执行查询选项显示计划按钮 SQL语句的结果查询数据显示查询 ...
第二十五章使用系统监视器 - 应用程序监视器
文章目录第二十五章使用系统监视器 - 配置健康监视器类设置运行状况监视器选项应用程序监视器应用程序监视器概述第二十五章使用系统监视器 - 配置健康监视器类此子菜单中的选项可让自定义 H ...
Linux的基本学习（十五）——认识系统服务
Linux的基本学习(十五)--认识系统服务前言继续学习Linux 什么是daemon与服务(service) 从CentOS 7.x开始,传统的init已经被抛弃,取而代之的是systemd 什 ...