Spring Batch 批处理（1） - 简介及使用场景

时间  2020-12-14
标签 java spring 数据库 设计模式 缓存 服务器 数据结构 多线程 架构 并发 栏目 Spring 繁體版
原文   原文链接

什么是 Spring Batch

 

介绍

Spring Batch 做为 Spring 的子项目，是一款基于 Spring 的企业批处理框架。经过它能够构建出健壮的企业批处理应用。Spring Batch 不只提供了统一的读写接口、丰富的任务处理方式、灵活的事务管理及并发处理，同时还支持日志、监控、任务重启与跳过等特性，大大简化了批处理应用开发，将开发人员从复杂的任务配置管理过程当中解放出来，使他们能够更多地去关注核心的业务处理过程。java

Spring Batch把批处理简化为Job和Job step两部分，在Job step中，把数据处理分为读数据(Reader)、处理数据(Processor)、写数据（Writer）三个步骤，异常处理机制分为跳过、重试、重启三种，做业方式分为多线程、并行、远程、分区四种。开发者在开发过程当中，大部分工做是根据业务要求编写Reader、Processor和Writer便可，提升了批处理开发的效率。同时Spring Batch自己也提供了不少默认的Reader和Writer，开箱即用。spring

官网详细介绍：https://spring.io/projects/spring-batch数据库

 

 

架构组件分类

 

  • Application(应用层):包含开发者应用Spring-batch编写的全部批处理做业和自定义代码;设计模式

  • Batch Core(核心层):包含加载和控制批处理做业所必需的核心类，它包含了Job,Step,JobLauncher的实现;缓存

  • Infrastructure(基础架构层)：基础架构层包含了Reader(ItemReader),Writer(ItemWriter),Services能够被应用层和核心层使用;服务器

 

file

 

 

优点

 

  • 丰富的开箱即用组件 开箱即用组件包括各类资源的读、写。读/写：支持文本文件读/写、XML文件读/写、数据库读/写、JMS队列读/写等。还提供做业仓库，做业调度器等基础设施，大大简化开发复杂度。数据结构

  • 面向chunk处理 支持屡次读、一次写、避免屡次对资源的写入，大幅提高批处理效率。多线程

  • 事务管理能力 默认采用Spring提供的声明式事务管理模型，面向Chunk的操做支持事务管理，同时支持为每一个tasklet操做设置细粒度的事务配置：隔离级别、传播行为、超时设置等。架构

  • 元数据管理 自动记录Job和Step的执行状况、包括成功、失败、失败的异常信息、执行次数、重试次数、跳过次数、执行时间等，方便后期的维护和查看。并发

  • 易监控的批处理应用 提供了灵活的监控模式，包括直接查看数据库、经过Spring Batch提供的API查看、JMX控制台查看等。其中还说到Spring Batch Admin，不过这个项目已不维护，改成用Spring Cloud Data Flow了。

  • 丰富的流程定义 支持顺序任务、条件分支任务、基于这两种任务能够组织复杂的任务流程。

  • 健壮的批处理应用 支持做业的跳过、重试、重启能力、避免因错误致使批处理做业的异常中断。

  • 易扩展的批处理应用 扩展机制包括多线程执行一个Step(Multithreaded step)、多线程并行执行多个Step(Parallelizing step)、远程执行做业（Remote chunking）、分区执行（partitioning step）。

  • 复用企业现有IT资产 提供多种Adapter能力，使得企业现有的服务能够方便集成到批处理应用中。避免从新开发、达到复用企业遗留的服务资产。

 

 

使用场景

 

  • 按期提交批处理任务

  • 并行批处理

  • 企业消息驱动处理

  • 大规模并行批处理

  • 失败后手动或定时重启

  • 按顺序处理依赖的任务（可扩展为工做流驱动的批处理）

  • 部分处理：跳过记录（例如，回滚时）

  • 批处理事务

 

 

原则与建议

 

当咱们构建一个批处理的过程时，必须注意如下原则：

一、一般状况下，批处理的过程对系统和架构的设计要够要求比较高，所以尽量的使用通用架构来处理批量数据处理，下降问题发生的可能性。Spring Batch是一个是一个轻量级的框架，适用于处理一些灵活并无到海量的数据。

二、批处理应该尽量的简单，尽可能避免在单个批处理中去执行过于复杂的任务。咱们能够将任务分红多个批处理或者多个步骤去实现。

三、保证数据处理和物理数据紧密相连。笼统的说就是咱们在处理数据的过程当中有不少步骤，在某些步骤执行完时应该就写入数据，而不是等全部都处理完。

四、尽量减小系统资源的使用、尤为是耗费大量资源的IO以及跨服务器引用，尽可能分配好数据处理的批次。

五、按期分析系统的IO使用状况、SQL语句的执行状况等，尽量的减小没必要要的IO操做。优化的原则有：

  • 尽可能在一次事物中对同一数据进行读取或写缓存。
  • 一次事物中，尽量在开始就读取全部须要使用的数据。
  • 优化索引，观察SQL的执行状况，尽可能使用主键索引，尽可能避免全表扫描或过多的索引扫描。
  • SQL中的where尽量经过主键查询。

六、不要在批处理中对相同的数据执行2次相同的操做。

七、对于批处理程序而言应该在批处理启动以前就分配足够的内存，以避免处理的过程当中去从新申请新的内存页。

八、对数据的完整性应该从最差的角度来考虑，每一步的处理都应该创建完备的数据校验。

九、对于数据的总量咱们应该有一个和数据记录在数据结构的某个字段 上。

十、全部的批处理系统都须要进行压力测试。

十一、若是整个批处理的过程是基于文件系统，在处理的过程当中请切记完成文件的备份以及文件内容的校验。


 

通用策略

 

和软件开发的设计模式同样，批处理也有各类各样的现成模式可供参考。当一个开发（设计）人员开始执行批处理任务时，应该将业务逻辑拆分为一下的步骤或者板块分批执行：

  • 数据转换：某个（某些）批处理的外部数据可能来自不一样的外部系统或者外部提供者，这些数据的结构千差万别。在统一进行批量数据处理以前须要对这些数据进行转换，合并为一个统一的结构。所以在数据开始真正的执行业务处理以前，先要使用其余的方法或者一些批处理任务将这些数据转换为统一的格式。

  • 数据校验：批处理是对大量数据进行处理，而且数据的来源千差万别，因此批处理的输入数据须要对数据的完整性性进行校验（好比校验字段数据是否缺失）。另外批处理输出的数据也须要进行合适的校验（例如处理了100条数据，校验100条数据是否校验成功）

  • 提取数据：批处理的工做是逐条从数据库或目标文件读取记录（records）,提取时能够经过一些规则从数据源中进行数据筛选。

  • 数据实时更新处理：根据业务要求，对实时数据进行处理。某些时候一行数据记录的处理须要绑定在一个事物之下。

  • 输出记录到标准的文档格式：数据处理完成以后须要根据格式写入到对应的外部数据系统中。

以上五个步骤是一个标准的数据批处理过程，Spring batch框架为业务实现提供了以上几个功能入口。


 

数据额外处理


某些状况须要实现对数据进行额外处理，在进入批处理以前经过其余方式将数据进行处理。主要内容有：

排序：因为批处理是以独立的行数据（record）进行处理的，在处理的时候并不知道记录先后关系。所以若是须要对总体数据进行排序，最好事先使用其余方式完成。

分割：数据拆分也建议使用独立的任务来完成。理由相似排序，由于批处理的过程都是以行记录为基本处理单位的，没法再对分割以后的数据进行扩展处理。

合并：理由如上。

 

 

Spring Batch核心概念

 

file

 

Spring Batch在基础架构层，把任务抽象为Job和Step，一个Job由多个Step来完成，step就是每一个job要执行的单个步骤。

一、Job:是一个接口，接口中定义了一个做业是怎么样执行的

二、JobInstance：是job的一次执行，一个JobInstance可重复执行，若是上一次执行失败下次执行的时候还会从新执行上次失败的job，每一次执行就是一个JobExceution

三、JobParameters:做为参数能够用来启动Job,而且能够用来标识不一样的Job，运行时提供给JobInstance,jonExceution根据状态和参数决定下一次是否继续执行

四、JobExceution：每一次尝试执行一个Job的时候，咱们就能够将其称为一个JobExceution，这个执行的结果能够为成功，也能够为失败，例如一个JobInstance执行失败了，下一次执行他传入的参数是上次执行的时间，他将会继续执行，这样始终执行的是一个JobInstance，而产生了两个JobExceution

五、Step:主要分为两块

（1）Tasklet：是一个任务单元，它是属于能够重复利用的东西。接口其中包含了一个惟一的方法execute();

（2）Chunk-based：chunk就是数据块，你须要定义多大的数据量是一个chunk。Chunk里面就是不断循环的一个流程，读数据，处理数据，而后写数据。Spring Batch会不断的循环这个流程，直到批处理数据完成。

  • ·itemReader：数据输入input：对于一个Step而言，每次读取一个条目；
  • ·itemProcessor:数据处理processing
  • ·ItemWriter：数据输出output：对于一个Step而言，每次根据设定输出批量一个条目；

六、StepExecution:一个Step的每一次尝试执行，都会建立一个StepExection,在一个Step实际开始执行的时候建立

七、ExecutionContext:执行上下文，表明的是一个key-value键值对的集合，能够被Spring框架进行在持久化管理，可以是开发人员存储持久化状态，每个JobExecution以及每个StepExecution的执行都会对应一个执行上下文（ExecutionContext）;对于StepExecution在每一次提交点时就会保存一下执行上下文，而对于Job是在每个StepExecution执行之间进行保存，例如，咱们从Step1换到Step2是就会保存;

八、JobLauncher：接口，用于启动和加载Job，根据传入的参数进行启动,返回Job一次执行的状况

九、JobRepository：Job及Job的运行结果和状态、Step的运行结果和状态，都会保存在JobRepository中。

 

file

 

 

概念说明可见下表：

领域对象 描述
JobRepository 做业仓库，保存Job、Step执行过程当中的状态及结果
JobLauncher 做业执行器，是执行Job的入口
Job 一个批处理任务，由一个或多个Step组成
Step 一个任务的具体的执行逻辑单位
Item 一条数据记录
ItemReader 从数据源读数据
ItemProcessor 对数据进行处理，如数据清洗、转换、过滤、校验等
ItemWriter 写入数据到指定目标
Chunk 给定数量的Item集合，如读取到chunk数量后，才进行写操做
Tasklet Step中具体执行逻辑，可重复执行

 

 

Spring Batch数据表

file

batch_job_instance：这张表能看到每次运行的job名字。

file

batch_job_execution：这张表能看到每次运行job的开始时间，结束时间，状态，以及失败后的错误消息是什么。

file

batch_step_execution：这张表你能看到更多关于step的详细信息。好比step的开始时间，结束时间，提交次数，读写次数，状态，以及失败后的错误信息等。 图片描述

 

 

Job


简单的说Job是封装一个批处理过程的实体，与其余的Spring项目相似，Job能够经过XML或Java类配置，称为“Job Configuration”。以下图Job是单个批处理的最顶层。


为了便于理解，能够简单的将Job理解为是每一步（Step）实例的容器。他结合了多个Step，为它们提供统一的服务同时也为Step提供个性化的服务，好比步骤重启。一般状况下Job的配置包含如下内容：

  • Job的名称
  • 定义和排序Step执行实例。
  • 标记每一个Step是否能够重启。

 

Spring Batch为Job接口提供了默认的实现——SimpleJob，其中实现了一些标准的批处理方法。下面的代码展现了如可注入一个Job。

@Bean
public Job testJob() {
    return this.jobBuilderFactory.get("testJob") //get中命名了Job的名称
                     .start(stepOne())  
                     .next(stepTwo())
                     .next(stepThree())
                     .end()
                     .build();
}

 

JobInstance


JobInstance是指批处理做业运行的实例。

例如一个批处理必须在天天执行一次，系统在2019年5月1日执行了一次咱们称之为2019-05-01的实例，相似的还会有2019-05-0二、2019-05-03实例。

一般状况下，一个JobInstance对应一个JobParameters，对应多个JobExecution。（JobParameters、JobExecution见后文）。同一个JobInstance具备相同的上下文（ExecutionContext内容见后文）。

 

JobParameters


前面讨论了JobInstance与Job的区别，可是具体的区别内容都是经过JobParameters体现的。一个JobParameters对象中包含了一系列Job运行相关的参数，这些参数能够用于参考或者用于实际的业务使用。对应的关系以下图：

file


当咱们执行2个不一样的JobInstance时JobParameters中的属性都会有差别。能够简单的认为一个JobInstance的标识就是Job+JobParameters。

 

JobExecution


JobExecution能够理解为单次运行Job的容器。一次JobInstance执行的结果多是成功、也多是失败。可是对于Spring Batch框架而言，只有返回运行成功才会视为完成一次批处理。

例如2019-05-01执行了一次JobInstance，可是执行的过程失败，所以第二次还会有一个“相同的”的JobInstance被执行。

Job用于定义批处理如何执行，JobInstance纯粹的就是一个处理对象，把全部的运行内容和信息组织在一块儿，主要是为了当面临问题时定义正确的重启参数。而JobExecution是运行时的“容器”，记录动态运行时的各类属性和上线文。


他包括的信息有：

属性 说明
status 状态类名为BatchStatus，它指示了执行的状态。在执行的过程当中状态为BatchStatus#STARTED，失败：BatchStatus#FAILED，完成：BatchStatus#COMPLETED
startTime java.util.Date对象，标记批处理任务启动的系统时间，批处理任务未启动数据为空
endTime java.util.Date对象，结束时间不管是否成功都包含该数据，如未处理完为空
exitStatus ExitStatus类，记录运行结果。
createTime java.util.Date,JobExecution的建立时间，某些使用execution已经建立可是并未开始运行。
lastUpdate java.util.Date，最后一次更新时间
executionContext 批处理任务执行的全部用户数据
failureExceptions 记录在执行Job时的异常，对于排查问题很是有用

 

以上这些内容Spring Batch都会经过JobRepository进行持久化（这些信息官方文成称之为MetaData），所以在对应的数据源中能够看到下列信息：

BATCH_JOB_INSTANCE：

JOB_INST_ID JOB_NAME
1 EndOfDayJob

 

BATCH_JOB_EXECUTION_PARAMS：

JOB_EXECUTION_ID TYPE_CD KEY_NAME DATE_VAL IDENTIFYING
1 DATE schedule.Date 2019-01-01 TRUE

 

BATCH_JOB_EXECUTION：

JOB_EXEC_ID JOB_INST_ID START_TIME END_TIME STATUS
1 1 2019-01-01 21:00 2017-01-01 21:30 FAILED

 

当某个Job批处理任务失败以后会在对应的数据库表中路对应的状态。假设1月1号执行的任务失败，技术团队花费了大量的时间解决这个问题到了次日才继续执行这个任务。

BATCH_JOB_INSTANCE：

JOB_INST_ID JOB_NAME
1 EndOfDayJob
2 EndOfDayJob

 

BATCH_JOB_EXECUTION_PARAMS：

JOB_EXECUTION_ID TYPE_CD KEY_NAME DATE_VAL IDENTIFYING
1 DATE schedule.Date 2019-01-01 TRUE
2 DATE schedule.Date 2019-01-01 TRUE
3 DATE schedule.Date 2019-01-02 TRUE

 

BATCH_JOB_EXECUTION：

JOB_EXEC_ID JOB_INST_ID START_TIME END_TIME STATUS
1 1 2019-01-01 21:00 2017-01-01 21:30 FAILED
2 1 2019-01-02 21:00 2017-01-02 21:30 COMPLETED
3 2 2019-01-02 21:31 2017-01-02 22:29 COMPLETED

 

从数据上看好似JobInstance是一个接一个顺序执行的，可是对于Spring Batch并无进行任何控制。不一样的JobInstance颇有多是同时在运行（相同的JobInstance同时运行会抛出JobExecutionAlreadyRunningException异常）。

 

 

Step


Step是批处理重复运行的最小单元，它按照顺序定义了一次执行的必要过程。

所以每一个Job能够视做由一个或多个多个Step组成。一个Step包含了全部全部进行批处理的必要信息，这些信息的内容是由开发人员决定的并无统一的标准。一个Step能够很简单，也能够很复杂。他能够是复杂业务的组合，也有可能仅仅用于迁移数据。

与JobExecution的概念相似，Step也有特定的StepExecution，关系结构以下：

file


 

StepExecution


StepExecution表示单次执行Step的容器，每次Step执行时都会有一个新的StepExecution被建立。与JobExecution不一样的是，当某个Step执行失败后默认并不会从新执行。StepExecution包含如下属性：

属性 说明
status 状态类名为BatchStatus，它指示了执行的状态。在执行的过程当中状态为BatchStatus#STARTED，失败：BatchStatus#FAILED，完成：BatchStatus#COMPLETED
startTime java.util.Date对象，标记StepExecution启动的系统时间，未启动数据为空
endTime java.util.Date对象，结束时间，不管是否成功都包含该数据，如未处理完为空
exitStatus ExitStatus类，记录运行结果。
createTime java.util.Date,JobExecution的建立时间，某些使用execution已经建立可是并未开始运行。
lastUpdate java.util.Date，最后一次更新时间
executionContext 批处理任务执行的全部用户数据
readCount 成功读取数据的次数
wirteCount 成功写入数据的次数
commitCount 成功提交数据的次数
rollbackCount 回归数据的次数，有业务代码触发
readSkipCount 当读数据发生错误时跳过处理的次数
processSkipCount 当处理过程发生错误，跳过处理的次数
filterCount 被过滤规则拦截未处理的次数
writeSkipCount 写数据失败，跳过处理的次数


 

ExecutionContext


前文已经屡次提到ExecutionContext。能够简单的认为ExecutionContext提供了一个Key/Value机制，在StepExecution和JobExecution对象的任何位置均可以获取到ExecutionContext中的任何数据。最有价值的做用是记录数据的执行位置，以便发生重启时候从对应的位置继续执行：

executionContext.putLong(getKey(LINES_READ_COUNT), reader.getPosition())


好比在任务中有一个名为“loadData”的Step，他的做用是从文件中读取数据写入到数据库，当第一次执行失败后，数据库中有以下数据：

BATCH_JOB_INSTANCE：

JOB_INST_ID JOB_NAME
1 EndOfDayJob

 

BATCH_JOB_EXECUTION_PARAMS：

JOB_INST_ID TYPE_CD KEY_NAME DATE_VAL
1 DATE schedule.Date 2019-01-01

 

BATCH_JOB_EXECUTION：

JOB_EXEC_ID JOB_INST_ID START_TIME END_TIME STATUS
1 1 2017-01-01 21:00 2017-01-01 21:30 FAILED

 

BATCH_STEP_EXECUTION：

STEP_EXEC_ID JOB_EXEC_ID STEP_NAME START_TIME END_TIME STATUS
1 1 loadData 2017-01-01 21:00 2017-01-01 21:30 FAILED

 

BATCH_STEP_EXECUTION_CONTEXT：

STEP_EXEC_ID SHORT_CONTEXT
1 {piece.count=40321}

 

在上面的例子中，Step运行30分钟处理了40321个“pieces”，咱们姑且认为“pieces”表示行间的行数（实际就是每一个Step完成循环处理的个数）。

这个值会在每一个commit以前被更新记录在ExecutionContext中（更新须要用到StepListener后文会详细说明）。

当咱们再次重启这个Job时并记录在BATCH_STEP_EXECUTION_CONTEXT中的数据会加载到ExecutionContext中,这样当咱们继续执行批处理任务时能够从上一次中断的位置继续处理。

例以下面的代码在ItemReader中检查上次执行的结果，并从中断的位置继续执行：

if (executionContext.containsKey(getKey(LINES_READ_COUNT))) {
    log.debug("Initializing for restart. Restart data is: " + executionContext);

    long lineCount = executionContext.getLong(getKey(LINES_READ_COUNT));

    LineReader reader = getReader();

    Object record = "";
    while (reader.getPosition() < lineCount && record != null) {
        record = readLine();
    }
}



ExecutionContext是根据JobInstance进行管理的，所以只要是相同的实例都会具有相同的ExecutionContext（不管是否中止）。此外经过如下方法均可以得到一个ExecutionContext：

ExecutionContext ecStep = stepExecution.getExecutionContext();
ExecutionContext ecJob = jobExecution.getExecutionContext();

 

可是这2个ExecutionContext并不相同，前者是在一个Step中每次Commit数据之间共享，后者是在Step与Step之间共享。


 

JobRepository


JobRepository是全部前面介绍的对象实例的持久化机制。他为JobLauncher、Job、Step的实现提供了CRUD操做。当一个Job第一次被启动时，一个JobExecution会从数据源中获取到，同时在执行的过程当中StepExecution、JobExecution的实现都会记录到数据源中。使用@EnableBatchProcessing注解后JobRepository会进行自动化配置。

 

JobLauncher


JobLauncher为Job的启动运行提供了一个边界的入口，在启动Job的同时还能够定制JobParameters：

public interface JobLauncher {
    public JobExecution run(Job job, JobParameters jobParameters) throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException, JobParametersInvalidException;
}

参考：

https://my.oschina.net/mianshenglee/blog/3058569

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程