文档章节

Spring Batch(4)——Item概念及使用代码

随风溜达的向日葵
 随风溜达的向日葵
发布于 07/08 21:23
字数 3612
阅读 835
收藏 10

批处理概念 中介绍一个标准的批处理分为 JobStep。本文将结合代码介绍在StepReaderProcessorWriter的实际使用。

Reader

Reader是指从各种各样的外部输入中获取数据,框架为获取各种类型的文件已经预定义了常规的Reader实现类。Reader通过ItemReader接口实现:

public interface ItemReader<T> {
    T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException;
}

read方法的作用就是读取一条数据,数据以泛型T的实体结构返回,当read返回null时表示所有数据读取完毕。返回的数据可以是任何结构,比如文件中的一行字符串,数据库的一行数据,或者xml文件中的一系列元素,只要是一个Java对象即可。

Writer

Writer通过ItemWriter接口实现:

public interface ItemWriter<T> {
    void write(List<? extends T> items) throws Exception;
}

WriterReader的反向操作,是将数据写入到特定的数据源中。在Step控制一文已经介绍Writer是根据chunk属性设定的值按列表进行操作的,所以传入的是一个List结构。chunk用于表示批处理的事物分片,因此需要注意的是,在writer方法中进行完整数据写入事物操作。例如向数据库写入List中的数据,在写入完成之后再提交事物。

读写的组合模式

无论是读还是写,有时会需要从多个不同的来源获取文件,或者写入到不同的数据源,或者是需要在读和写之间处理一些业务。可以使用组合模式来实现这个目的:

public class CompositeItemWriter<T> implements ItemWriter<T> {
    ItemWriter<T> itemWriter;
    public CompositeItemWriter(ItemWriter<T> itemWriter) {
        this.itemWriter = itemWriter;
    }

    public void write(List<? extends T> items) throws Exception {
        //Add business logic here
       itemWriter.write(items);
    }

    public void setDelegate(ItemWriter<T> itemWriter){
        this.itemWriter = itemWriter;
    }
}

Processor

除了使用组合模式,直接使用Processor是一种更优雅的方法。ProcessorStep中的可选项,但是批处理大部分时候都需要对数据进行处理,因此框架提供了ItemProcessor接口来满足Processor过程:

public interface ItemProcessor<I, O> {
    O process(I item) throws Exception;
}

Processor的结构非常简单也是否易于理解。传入一个类型I,然后由Processor处理成为O。

Processor链

在一个Step中可以使用多个Processor来按照顺序处理业务,此时同样可以使用CompositeItem模式来实现:


@Bean
public CompositeItemProcessor compositeProcessor() {
    //创建 CompositeItemProcessor
    CompositeItemProcessor<Foo,Foobar> compositeProcessor = new CompositeItemProcessor<Foo,Foobar>();
    List itemProcessors = new ArrayList();
    //添加第一个 Processor
    itemProcessors.add(new FooTransformer());
    //添加第二个 Processor
    itemProcessors.add(new BarTransformer());
    //添加链表
    compositeProcessor.setDelegates(itemProcessors);
    return processor;
}

过滤记录

Reader读取数据的过程中,并不是所有的数据都可以使用,此时Processor还可以用于过滤非必要的数据,同时不会影响Step的处理过程。只要ItemProcesspr的实现类在procss方法中返回null即表示改行数据被过滤掉了。

ItemStream

Step控制一文中已经提到了ItemStream。在数据批处理概念中提到过,Spring Batch的每一步都是无状态的,进而ReaderWriter也是无状态的,这种方式能够很好的隔离每行数据的处理,也能将容错的范围收窄到可以空子的范围。但是这并不意味着整个批处理的过程中并不需要控制状态。例如从数据库持续读入或写入数据,每次ReaderWriter都单独去申请数据源的链接、维护数据源的状态(打开、关闭等)。因此框架提供了ItemStream接口来完善这些操作:

public interface ItemStream {
    void open(ExecutionContext executionContext) throws ItemStreamException;
    void update(ExecutionContext executionContext) throws ItemStreamException;
    void close() throws ItemStreamException;
}

持久化数据

在使用Spring Batch之前需要初始化他的元数据存储(Meta-Data Schema),也就是要将需要用到的表导入到对应的数据库中。当然,Spring Batch支持不使用任何持久化数据库,仅仅将数据放到内存中,不设置DataSource即可。

初始化序列

Spring Batch相关的工作需要使用序列SEQUENCE

CREATE SEQUENCE BATCH_STEP_EXECUTION_SEQ;
CREATE SEQUENCE BATCH_JOB_EXECUTION_SEQ;
CREATE SEQUENCE BATCH_JOB_SEQ;

有些数据库不支持SEQUENCE,可以通过表代理,比如在MySql(InnoDB数据库)中:

CREATE TABLE BATCH_STEP_EXECUTION_SEQ (ID BIGINT NOT NULL);
INSERT INTO BATCH_STEP_EXECUTION_SEQ values(0);
CREATE TABLE BATCH_JOB_EXECUTION_SEQ (ID BIGINT NOT NULL);
INSERT INTO BATCH_JOB_EXECUTION_SEQ values(0);
CREATE TABLE BATCH_JOB_SEQ (ID BIGINT NOT NULL);
INSERT INTO BATCH_JOB_SEQ values(0);

关于Version字段

某些表中都有Version字段。因为Spring的更新策略是乐观锁,因此在进行数据更新之后都会对表的Version字段进行+1处理。在内存与数据库交互的过程中,会使用采用getVersionincreaseVersion(+1)updateDataAndVersion的过程,如果在update的时候发现Version不是预计的数值(+1),则会抛出OptimisticLockingFailureException的异常。当同一个Job在进群中不同服务上执行时,需要注意这个问题。

BATCH_JOB_INSTANCE

BATCH_JOB_INSTANCE用于记录JobInstance,在数据批处理概念中介绍了他的工作方式,其结构为:

CREATE TABLE BATCH_JOB_INSTANCE  (
  JOB_INSTANCE_ID BIGINT  PRIMARY KEY ,
  VERSION BIGINT,
  JOB_NAME VARCHAR(100) NOT NULL ,
  JOB_KEY VARCHAR(2500)
);
字段说明
JOB_INSTANCE_ID主键,主键与单个JobInstance相关。当获取到某个JobInstance实例后,通过getId方法可以获取到此数据
VERSION
JOB_NAMEJob的名称,用于标记运行的Job,在创建Job时候指定
JOB_KEYJobParameters的序列化数值。在数据批处理概念中介绍了一个JobInstance相当于Job+JobParameters。他用于标记同一个Job不同的实例

BATCH_JOB_EXECUTION_PARAMS

BATCH_JOB_EXECUTION_PARAMS对应的是JobParameters对象。其核心功能是存储Key-Value结构的各种状态数值。字段中IDENTIFYING=true用于标记那些运行过程中必须的数据(可以理解是框架需要用到的数据),为了存储key-value结构该表一个列数据格式:

CREATE TABLE BATCH_JOB_EXECUTION_PARAMS  (
	JOB_EXECUTION_ID BIGINT NOT NULL ,
	TYPE_CD VARCHAR(6) NOT NULL ,
	KEY_NAME VARCHAR(100) NOT NULL ,
	STRING_VAL VARCHAR(250) ,
	DATE_VAL DATETIME DEFAULT NULL ,
	LONG_VAL BIGINT ,
	DOUBLE_VAL DOUBLE PRECISION ,
	IDENTIFYING CHAR(1) NOT NULL ,
	constraint JOB_EXEC_PARAMS_FK foreign key (JOB_EXECUTION_ID)
	references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
);
字段说明
JOB_EXECUTION_IDBATCH_JOB_EXECUTION表关联的外键,详见数据批处理概念JobJobInstanceJobExecute的关系
TYPE_CD用于标记数据的对象类型,例如 stringdatelongdouble,非空
KEY_NAMEkey的值
STRING_VALstring类型的数值
DATE_VALdate类型的数值
LONG_VALlong类型的数值
DOUBLE_VALdouble类型的数值
IDENTIFYING标记这对key-valuse是否来自于JobInstace自身

BATCH_JOB_EXECUTION

关联JobExecution,每当运行一个Job都会产生一个新的JobExecution,对应的在表中都会新增一行数据。

CREATE TABLE BATCH_JOB_EXECUTION  (
  JOB_EXECUTION_ID BIGINT  PRIMARY KEY ,
  VERSION BIGINT,
  JOB_INSTANCE_ID BIGINT NOT NULL,
  CREATE_TIME TIMESTAMP NOT NULL,
  START_TIME TIMESTAMP DEFAULT NULL,
  END_TIME TIMESTAMP DEFAULT NULL,
  STATUS VARCHAR(10),
  EXIT_CODE VARCHAR(20),
  EXIT_MESSAGE VARCHAR(2500),
  LAST_UPDATED TIMESTAMP,
  JOB_CONFIGURATION_LOCATION VARCHAR(2500) NULL,
  constraint JOB_INSTANCE_EXECUTION_FK foreign key (JOB_INSTANCE_ID)
  references BATCH_JOB_INSTANCE(JOB_INSTANCE_ID)
) ;
字段说明
JOB_EXECUTION_IDJobExecution的主键,JobExecution::getId方法可以获取到该值
VERSION
JOB_INSTANCE_ID关联到JobInstace的外键,详见数据批处理概念JobJobInstanceJobExecute的关系
CREATE_TIME创建时间戳
START_TIME开始时间戳
END_TIME结束时间戳,无论成功或失败都会更新这一项数据。如果某行数据该值为空表示运行期间出现错误,并且框架无法更新该值
STATUSJobExecute的运行状态:COMPLETEDSTARTED或者其他状态。此数值对应Java中BatchStatus枚举值
EXIT_CODEJobExecute执行完毕之后的退出返回值
EXIT_MESSAGEJobExecute退出的详细内容,如果是异常退出可能会包括异常堆栈的内容
LAST_UPDATED最后一次更新的时间戳

BATCH_STEP_EXECUTION

该表对应的是StepExecution,其结构和BATCH_JOB_EXECUTION基本相似,只是对应的对象是Step,增加了与之相对的一些字段数值:

CREATE TABLE BATCH_STEP_EXECUTION  (
  STEP_EXECUTION_ID BIGINT  PRIMARY KEY ,
  VERSION BIGINT NOT NULL,
  STEP_NAME VARCHAR(100) NOT NULL,
  JOB_EXECUTION_ID BIGINT NOT NULL,
  START_TIME TIMESTAMP NOT NULL ,
  END_TIME TIMESTAMP DEFAULT NULL,
  STATUS VARCHAR(10),
  COMMIT_COUNT BIGINT ,
  READ_COUNT BIGINT ,
  FILTER_COUNT BIGINT ,
  WRITE_COUNT BIGINT ,
  READ_SKIP_COUNT BIGINT ,
  WRITE_SKIP_COUNT BIGINT ,
  PROCESS_SKIP_COUNT BIGINT ,
  ROLLBACK_COUNT BIGINT ,
  EXIT_CODE VARCHAR(20) ,
  EXIT_MESSAGE VARCHAR(2500) ,
  LAST_UPDATED TIMESTAMP,
  constraint JOB_EXECUTION_STEP_FK foreign key (JOB_EXECUTION_ID)
  references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
) ;

未填入内容部分见BATCH_JOB_EXECUTION说明。

字段说明
STEP_EXECUTION_IDStepExecute对应的主键
VERSION
STEP_NAMEStep名称
JOB_EXECUTION_ID关联到BATCH_JOB_EXECUTION表的外键,标记该StepExecute所属的JobExecute
START_TIME
END_TIME
STATUS
COMMIT_COUNT执行过程中,事物提交的次数,该值与数据的规模以及chunk的设置有关
READ_COUNT读取数据的次数
FILTER_COUNTProcessor中过滤记录的次数
WRITE_COUNT吸入数据的次数
READ_SKIP_COUNT读数据的跳过次数
WRITE_SKIP_COUNT写数据的跳过次数
PROCESS_SKIP_COUNTProcessor跳过的次数
ROLLBACK_COUNT回滚的次数
EXIT_CODE
EXIT_MESSAGE
LAST_UPDATED

BATCH_JOB_EXECUTION_CONTEXT

该表会记录所有与Job相关的ExecutionContext信息。每个ExecutionContext都对应一个JobExecution,在运行的过程中它包含了所有Job范畴的状态数据,这些数据在执行失败后对于后续处理有中重大意义。

CREATE TABLE BATCH_JOB_EXECUTION_CONTEXT  (
  JOB_EXECUTION_ID BIGINT PRIMARY KEY,
  SHORT_CONTEXT VARCHAR(2500) NOT NULL,
  SERIALIZED_CONTEXT CLOB,
  constraint JOB_EXEC_CTX_FK foreign key (JOB_EXECUTION_ID)
  references BATCH_JOB_EXECUTION(JOB_EXECUTION_ID)
) ;
字段说明
JOB_EXECUTION_ID关联到JobExecution的外键,建立JobExecutionExecutionContext的关系。
SHORT_CONTEXT标记SERIALIZED_CONTEXT的版本号
SERIALIZED_CONTEXT序列化的ExecutionContext

BATCH_STEP_EXECUTION_CONTEXT

StepExecutionContext相关的数据表,结构与BATCH_JOB_EXECUTION_CONTEXT完全一样。

表索引建议

上面的所有建表语句都没有提供索引,但是并不代表索引没有价值。当感觉到SQL语句的执行有效率问题时候,可以增加索引。

索引带来的价值取决于SQL查询的频率以及关联关系,下面是Spring Batch框架在运行过程中会用到的一些查询条件语句,用于参考优化索引:

Where条件执行频率
BATCH_JOB_INSTANCEJOB_NAME = ? and JOB_KEY = ?每次Job启动执时
BATCH_JOB_EXECUTIONJOB_INSTANCE_ID = ?每次Job重启时
BATCH_EXECUTION_CONTEXTEXECUTION_ID = ? and KEY_NAME = ?chunk的大小而定
BATCH_STEP_EXECUTIONVERSION = ?chunk的大小而定
BATCH_STEP_EXECUTIONSTEP_NAME = ? and JOB_EXECUTION_ID = ?每一个Step执行之前

使用案例

下面是Spring Batch一些简单的应用,源码在下列地址的simple工程:

Spring Batch提供了2种执行方式:命令行方式或Java内嵌方式。命令行方式是直到需要执行批处理任务的时候才启动程序,内嵌方式是结合Web工程或其他外部化框架来使用。2者最大的差别就是是否直接向IoCs注入一个Job实例。

通用基本配置

两种方式的基本配置都是一样的,通过ReaderProcessorWriter来组装一个Step。代码中Item并不涉及文件或数据库的操作,只是简单的模拟数据读取、处理、写入的过程。实体RecordMsg用于模拟数据转换,基本配置如下:

public class BatchDefaultConfig {
	@Bean
	//配置Step
	public Step simpleStep(StepBuilderFactory builder, ItemReader<Record> reader, ItemProcessor<Record, Msg> processor,
			ItemWriter<Msg> writer) {
		return builder.get("SimpleStep").<Record, Msg>chunk(10).reader(reader).processor(processor).writer(writer)
				.build();
	}

	@Bean
	//配置 Reader
	public ItemReader<Record> reader() {
		return new ItemReader<Record>() {
			private int count = 0;
			public Record read()
					throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
				return ++this.count < 100 ? new Record().setId(this.count).setMsg("Read Number:" + this.count) : null;
			}
		};
	}

	@Bean
	//配置 Processor
	public ItemProcessor<Record, Msg> processor() {
		return new ItemProcessor<Record, Msg>() {
			public Msg process(Record item) throws Exception {
				return new Msg("MSG GET INFO = " + item.getMsg());
			}
		};
	}

	@Bean
	//配置 Writer
	public ItemWriter<Msg> writer() {
		return new ItemWriter<Msg>() {
			private int batchCount = 0;
			public void write(List<? extends Msg> items) throws Exception {
				System.out.println("Batch Count : " + ++batchCount + ". Data:");
				for (Msg msg : items) {
					System.out.println(msg.getInfo());
				}
			}
		};
	}
}

命令行方式运行

有了基本配置之后,命令行运行的方式仅仅是向容器添加一个Job

@Configuration
//导入依赖配置
@Import({ BatchDefaultConfig.class })
public class BatchCommondConfig {
	@Bean
	public Job simpleJob(Step step, JobBuilderFactory builder) {
		return builder.get("SimpleJob").start(step).build(); //向容器返回一个Job的Bean
	}
}

然后启动Spring Framework则会自动启用Command Runner运行方式运行——先调用SpringApplication::callRunner方法,然后使用JobLauncherCommandLineRunner::execute运行:

public class CommondSample {
	public static void main(String[] args) throws DuplicateJobException {
		//模拟测试参数, 这些参数值在执行Java时从外部传入的,比如-Dkey=value
		String[] argsExt = new String[2];
		argsExt[0] = "BuilderParam1=Value1";
		argsExt[1] = "BuilderParam2=Value2";
		//运行Spring Framework
		SpringApplication.run(CommondSample.class, argsExt);
	}
}

启用之后观察数据库已经发生了变更。使用命令行需要通过 Java运行参数(-Dkey=value)传递JobParameters的数据,上面的代码模拟实现了相关的过程。

Java内嵌运行

Java内嵌的方式主要是用于搭配外部工程化使用,比如使用Web框架或则统一调度平台管之类的结构化框架来统一管理批处理任务。与命令行执行最大的区别就是不向容器注入Job

@Configuration
//导入进出配置 
@Import({BatchDefaultConfig.class})
public class BatchOperatoConfig {
	@Bean
	//返回JobFactory
	public JobFactory simpleJob(Step step, JobBuilderFactory builder) {
		SimpleJobFactory sampleJobFactory = new SimpleJobFactory();
		sampleJobFactory.setJob(builder.get("SimpleJob").start(step).build());
		return sampleJobFactory;
	}
}

配置代码向容器添加了一个JobFactory的实现类,JobFactory的两个接口一个是获取Job一个是获取Job的名称,SimpleJobFactory实现了JobFactory

public class SimpleJobFactory implements JobFactory {
	private Job job;
	public void setJob(Job job) {
		this.job = job;
	}
	@Override
	public Job createJob() {
		return job;
	}
	@Override
	public String getJobName() {
		return job.getName();
	}
}

最后通过SimpleJobFactory来启动一个Job

@SpringBootApplication
@EnableBatchProcessing
@EnableScheduling
public class OperatorSample {
	public static void main(String[] args) throws DuplicateJobException {
		new SuspendThread().run(); //挂起系统一直运行
		ConfigurableApplicationContext ctx = SpringApplication.run(OperatorSample.class);
		Cron cron = ctx.getBean(Cron.class);
		cron.register(); //注册JobFactory
		cron.runJobLaunch();
	}
}

@Service
class Cron {
	@Autowired
	JobLauncher jobLauncher;

	@Autowired
	private JobOperator jobOperator;

	@Autowired
	private JobRegistry jobRegistry;

	@Autowired
	private JobFactory jobFactory;

	//注册JobFactory
	void register() throws DuplicateJobException {
		jobRegistry.register(jobFactory);
	}

	//使用JobLaunch执行
	void runJobLaunch() {
		Map<String, JobParameter> map = new HashMap<>();
		map.put("Builder", new JobParameter("1"));
		map.put("Timer", new JobParameter("2"));
		jobLauncher.run(jobFactory.createJob(), new JobParameters(map));
	}

	@Scheduled(cron = "30 * * * * ? ")
	void task1() {
		System.out.println("1");
		runOperator();
	}

	//定时任务使用 JobOperator执行
	private void runOperator() {
		jobOperator.start("SimpleJob", "Builder=1,Timer=2");
	}
}

这里使用了2种执行方式:JobLauncherJobOperatorJobLauncher简单明了的启动一个批处理任务。而JobOperator扩展了一些用于Job管理的接口方法,观察JobOperator的源码可以发现它提供了获取ExecuteContext、检查JobInstance等功能,如果需要定制开发一个基于Web或者JMX管理批处理任务的系统,JobOperator更合适。JobOperator的第二个参数用于传递JobParameters,等号两端分别是keyvalue,逗号用于分割多行数据。

Job配置与运行提及过一个JobInstance相当于Job+JobParameters,因此虽然上面的代码使用了两种不同的运行方式,但是JobJobParameters是一样的。在运行被定时任务包裹的runOperator方法时,会一直抛出JobInstanceAlreadyExistsException异常,因为同一个实例不能运行2次。如果运行失败可以使用对应的restart方法。

后续会介绍各种ReaderWriter的使用。

© 著作权归作者所有

随风溜达的向日葵
粉丝 298
博文 91
码字总数 207714
作品 0
广州
其他
私信 提问
Spring Batch(5)——文件读写

在Spring batch由上至下的结构中Job、Step都是属于框架级别的的功能,大部分时候都是提供一些配置选项给开发人员使用,而Item中的、和是属于业务级别的,它开放了一些业务切入的接口。 但是文...

随风溜达的向日葵
07/09
0
0
Spring Batch(3)——Step控制

批处理任务的主要业务逻辑都是在中去完成的。可以将理解为运行的框架,而理解为业务功能。 Step配置 是中的工作单元,每一个涵盖了单行记录的处理闭环。下图是一个的简要结构: 一个通常涵盖...

随风溜达的向日葵
07/05
0
0
Spring Batch(2)——Job配置与运行

在 Spring Batch(1)——数据批处理概念——数据批处理概念") 文中介绍了批处理的概念以及Spring Batch相关的使用场景,后续将会陆续说明在代码层面如何使用。 引入 Spring batch的引入非常简...

随风溜达的向日葵
07/04
0
0
Spring Batch(6)——数据库批数据读写

前序文章陆续介绍了批处理的基本概念,Job使用、Step控制、Item的结构以及扁平文件的读写。本文将接着前面的内容说明数据库如何进行批处理读写。 数据读取 数据库是绝大部分系统要用到的数据...

随风溜达的向日葵
07/10
0
0
Spring Batch:简化批处理和离线处理的开发

作者 Ryan Slobojan 译者 宋玮 发布于 2008年7月4日 上午1时17分 转自 http://www.infoq.com/cn/news/2008/07/spring-batch-zh Spring Batch项目是一个轻量级的全面的基于Spring的批处理框架...

feewo
2010/10/20
1K
0

没有更多内容

加载失败,请刷新页面

加载更多

Linux输入法fcitx的安装问题

Fcitx 总共要安装的包如下 fcitxfcitx-binfcitx-config-commonfcitx-config-gtk | fcitx-config-gtk2fcitx-datafcitx-frontend-allfcitx-frontend-gtk2fcitx-frontend-gtk3......

CHONGCHEN
57分钟前
4
0
网络基础

前言: 最近整理一些以前的学习笔记(有部分缺失,会有些乱,日后再补)。 过去都是存储在本地,此次传到网络留待备用。 计算机网络的功能: 1.数据通信; 2.资源共享; 3.增加数据可靠性; 4....

迷失De挣扎
58分钟前
7
0
spring boot升级到spring cloud

1、先升级spring boot 版本到2.1.3 <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-dependencies</artifactId> <version>2.1.3.RELEAS......

moon888
今天
12
0
从蓝鲸视角谈DevOps

DevOps源于Development和Operations的组合 常见的定义 DevOps是一种重视“软件开发人员(Dev)”和“IT运维技术人员(Ops)”之间沟通合作的文化、运动或惯例。透过自动化“软件交付”和“架构变...

嘉为科技
今天
1
0
微服务设计 笔记

微服务设计 一、微服务架构理论 1.六边形架构 1)六边形架构(Hexagonal Architecture),又称为端口和适配器架构风格;使用适配器与外界进行交互,外界通过应用层API与内部进行交互。 2)经典...

啃不动地大坚果
今天
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部