关于java:Spring-Batch-批处理框架真心强啊

spring batch 简介

spring batch 是 spring 提供的一个数据处理框架。企业域中的许多应用程序须要批量解决能力在要害工作环境中执行业务操作。这些业务经营包含：

• 无需用户交互即可最无效地解决大量信息的自动化，简单解决。这些操作通常包含基于工夫的事件（例如月末计算，告诉或通信）。
• 在十分大的数据集中反复解决简单业务规定的定期利用（例如，保险利益确定或费率调整）。
• 集成从外部和内部零碎接管的信息，这些信息通常须要以事务形式格式化，验证和解决到记录零碎中。批处理用于每天为企业解决数十亿的交易。

Spring Batch 是一个轻量级，全面的批处理框架，旨在开发对企业零碎日常经营至关重要的弱小批处理应用程序。Spring Batch 构建了人们冀望的 Spring Framework 个性（生产力，基于 POJO 的开发方法和个别易用性），同时使开发人员能够在必要时轻松拜访和利用更高级的企业服务。Spring Batch 不是一个 schuedling 的框架。

Spring Batch 提供了可重用的性能，这些性能对于解决大量的数据至关重要，包含记录 / 跟踪，事务管理，作业处理统计，作业重启，跳过和资源管理。它还提供更高级的技术服务和性能，通过优化和分区技术实现极高容量和高性能的批处理作业。

Spring Batch 可用于两种简略的用例（例如将文件读入数据库或运行存储过程）以及简单的大量用例（例如在数据库之间挪动大量数据，转换它等等）上）。大批量批处理作业能够高度可扩大的形式利用该框架来解决大量信息。

Spring Batch 架构介绍

一个典型的批处理应用程序大抵如下：

• 从数据库，文件或队列中读取大量记录。
• 以某种形式解决数据。
• 以批改之后的模式写回数据。

其对应的示意图如下：

spring batch 的一个总体的架构如下：

在 spring batch 中一个 job 能够定义很多的步骤 step，在每一个 step 外面能够定义其专属的 ItemReader 用于读取数据，ItemProcesseor 用于解决数据，ItemWriter 用于写数据，而每一个定义的 job 则都在 JobRepository 外面，咱们能够通过 JobLauncher 来启动某一个 job。

Spring Batch 外围概念介绍

上面是一些概念是 Spring batch 框架中的外围概念。

什么是 Job

Job 和 Step 是 spring batch 执行批处理工作最为外围的两个概念。

其中 Job 是一个封装整个批处理过程的一个概念。Job 在 spring batch 的体系当中只是一个最顶层的一个抽象概念，体现在代码当中则它只是一个最上层的接口，其代码如下:

/**
 * Batch domain object representing a job. Job is an explicit abstraction
 * representing the configuration of a job specified by a developer. It should
 * be noted that restart policy is applied to the job as a whole and not to a
 * step.
 */
public interface Job {String getName();
 
 
 boolean isRestartable();
 
 
 void execute(JobExecution execution);
 
 
 JobParametersIncrementer getJobParametersIncrementer();
 
 
 JobParametersValidator getJobParametersValidator();}

在 Job 这个接口当中定义了五个办法，它的实现类次要有两种类型的 job，一个是 simplejob，另一个是 flowjob。在 spring batch 当中，job 是最顶层的形象，除 job 之外咱们还有 JobInstance 以及 JobExecution 这两个更加底层的形象。

一个 job 是咱们运行的根本单位，它外部由 step 组成。job 实质上能够看成 step 的一个容器。一个 job 能够依照指定的逻辑程序组合 step，并提供了咱们给所有 step 设置雷同属性的办法，例如一些事件监听，跳过策略。

Spring Batch 以 SimpleJob 类的模式提供了 Job 接口的默认简略实现，它在 Job 之上创立了一些规范性能。一个应用 java config 的例子代码如下：

@Bean
public Job footballJob() {return this.jobBuilderFactory.get("footballJob")
                     .start(playerLoad())
                     .next(gameLoad())
                     .next(playerSummarization())
                     .end()
                     .build();}

这个配置的意思是：首先给这个 job 起了一个名字叫 footballJob，接着指定了这个 job 的三个 step，他们别离由办法，playerLoad,gameLoad, playerSummarization 实现。

什么是 JobInstance

咱们在上文曾经提到了 JobInstance，他是 Job 的更加底层的一个形象，他的定义如下：

public interface JobInstance {
 /**
  * Get unique id for this JobInstance.
  * @return instance id
  */
 public long getInstanceId();
 /**
  * Get job name.
  * @return value of 'id' attribute from <job>
  */
 public String getJobName();}

他的办法很简略，一个是返回 Job 的 id，另一个是返回 Job 的名字。

JobInstance 指的是 job 运行当中，作业执行过程当中的概念。Instance 本就是实例的意思。

比如说当初有一个批处理的 job，它的性能是在一天完结时执行行一次。咱们假设这个批处理 job 的名字为 ’EndOfDay’。在这个状况下，那么每天就会有一个逻辑意义上的 JobInstance, 而咱们必须记录 job 的每次运行的状况。

什么是 JobParameters

在上文当中咱们提到了，同一个 job 每天运行一次的话，那么每天都有一个 jobIntsance，但他们的 job 定义都是一样的，那么咱们怎么来区别一个 job 的不同 jobinstance 了。无妨先做个猜测，尽管 jobinstance 的 job 定义一样，然而他们有的货色就不一样，例如运行工夫。

spring batch 中提供的用来标识一个 jobinstance 的货色是：JobParameters。JobParameters 对象蕴含一组用于启动批处理作业的参数，它能够在运行期间用于辨认或甚至用作参考数据。咱们假如的运行工夫，就能够作为一个 JobParameters。

例如, 咱们后面的 ’EndOfDay’ 的 job 当初曾经有了两个实例，一个产生于 1 月 1 日，另一个产生于 1 月 2 日，那么咱们就能够定义两个 JobParameter 对象：一个的参数是 01-01, 另一个的参数是 01-02。因而，辨认一个 JobInstance 的办法能够定义为：

因而，我么能够通过 Jobparameter 来操作正确的 JobInstance

什么是 JobExecution

JobExecution 指的是单次尝试运行一个咱们定义好的 Job 的代码层面的概念。job 的一次执行可能以失败也可能胜利。只有当执行胜利实现时，给定的与执行绝对应的 JobInstance 才也被视为实现。

还是以后面形容的 EndOfDay 的 job 作为示例，假如第一次运行 01-01-2019 的 JobInstance 后果是失败。那么此时如果应用与第一次运行雷同的 Jobparameter 参数（即 01-01-2019）作业参数再次运行，那么就会创立一个对应于之前 jobInstance 的一个新的 JobExecution 实例,JobInstance 依然只有一个。

JobExecution 的接口定义如下：

public interface JobExecution {
 /**
  * Get unique id for this JobExecution.
  * @return execution id
  */
 public long getExecutionId();
 /**
  * Get job name.
  * @return value of 'id' attribute from <job>
  */
 public String getJobName(); 
 /**
  * Get batch status of this execution.
  * @return batch status value.
  */
 public BatchStatus getBatchStatus();
 /**
  * Get time execution entered STARTED status. 
  * @return date (time)
  */
 public Date getStartTime();
 /**
  * Get time execution entered end status: COMPLETED, STOPPED, FAILED 
  * @return date (time)
  */
 public Date getEndTime();
 /**
  * Get execution exit status.
  * @return exit status.
  */
 public String getExitStatus();
 /**
  * Get time execution was created.
  * @return date (time)
  */
 public Date getCreateTime();
 /**
  * Get time execution was last updated updated.
  * @return date (time)
  */
 public Date getLastUpdatedTime();
 /**
  * Get job parameters for this execution.
  * @return job parameters  
  */
 public Properties getJobParameters();}

每一个办法的正文曾经解释的很分明，这里不再多做解释。只提一下 BatchStatus，JobExecution 当中提供了一个办法 getBatchStatus 用于获取一个 job 某一次顺便执行的一个状态。BatchStatus 是一个代表 job 状态的枚举类，其定义如下：

public enum BatchStatus {STARTING, STARTED, STOPPING, 
   STOPPED, FAILED, COMPLETED, ABANDONED }

这些属性对于一个 job 的执行来说是十分要害的信息，并且 spring batch 会将他们长久到数据库当中. 在应用 Spring batch 的过程当中 spring batch 会主动创立一些表用于存储一些 job 相干的信息，用于存储 JobExecution 的表为 batch_job_execution, 上面是一个从数据库当中截图的实例：

什么是 Step

每一个 Step 对象都封装了批处理作业的一个独立的阶段。事实上，每一个 Job 实质上都是由一个或多个步骤组成。每一个 step 蕴含定义和管制理论批处理所需的所有信息。任何特定的内容都由编写 Job 的开发人员自行决定。

一个 step 能够非常简单也能够非常复杂。例如，一个 step 的性能是将文件中的数据加载到数据库中，那么基于当初 spring batch 的反对则简直不须要写代码。更简单的 step 可能具备简单的业务逻辑，这些逻辑作为解决的一部分。

与 Job 一样，Step 具备与 JobExecution 相似的 StepExecution，如下图所示：

什么是 StepExecution

StepExecution 示意一次执行 Step, 每次运行一个 Step 时都会创立一个新的 StepExecution，相似于 JobExecution。然而，某个步骤可能因为其之前的步骤失败而无奈执行。且仅当 Step 理论启动时才会创立 StepExecution。

一次 step 执行的实例由 StepExecution 类的对象示意。每个 StepExecution 都蕴含对其相应步骤的援用以及 JobExecution 和事务相干的数据，例如提交和回滚计数以及开始和完结工夫。

此外，每个步骤执行都蕴含一个 ExecutionContext，其中蕴含开发人员须要在批处理运行中保留的任何数据，例如重新启动所需的统计信息或状态信息。上面是一个从数据库当中截图的实例：

什么是 ExecutionContext

ExecutionContext 即每一个 StepExecution 的执行环境。它蕴含一系列的键值对。咱们能够用如下代码获取 ExecutionContext

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

什么是 JobRepository

JobRepository 是一个用于将上述 job，step 等概念进行长久化的一个类。它同时给 Job 和 Step 以及下文会提到的 JobLauncher 实现提供 CRUD 操作。

首次启动 Job 时，将从 repository 中获取 JobExecution，并且在执行批处理的过程中，StepExecution 和 JobExecution 将被存储到 repository 当中。

@EnableBatchProcessing 注解能够为 JobRepository 提供主动配置。

什么是 JobLauncher

JobLauncher 这个接口的性能非常简单，它是用于启动指定了 JobParameters 的 Job，为什么这里要强调指定了 JobParameter，起因其实咱们在后面曾经提到了，jobparameter 和 job 一起能力组成一次 job 的执行。上面是代码实例：

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

下面 run 办法实现的性能是依据传入的 job 以及 jobparamaters 从 JobRepository 获取一个 JobExecution 并执行 Job。

什么是 Item Reader

ItemReader 是一个读数据的形象，它的性能是为每一个 Step 提供数据输出。当 ItemReader 以及读完所有数据时，它会返回 null 来通知后续操作数据曾经读完。Spring Batch 为 ItemReader 提供了十分多的有用的实现类，比方 JdbcPagingItemReader，JdbcCursorItemReader 等等。

ItemReader 反对的读入的数据源也是十分丰盛的，包含各种类型的数据库，文件，数据流，等等。简直涵盖了咱们的所有场景。

上面是一个 JdbcPagingItemReader 的例子代码：

@Bean
public JdbcPagingItemReader itemReader(DataSource dataSource, PagingQueryProvider queryProvider) {Map<String, Object> parameterValues = new HashMap<>();
        parameterValues.put("status", "NEW");
 
        return new JdbcPagingItemReaderBuilder<CustomerCredit>()
                                           .name("creditReader")
                                           .dataSource(dataSource)
                                           .queryProvider(queryProvider)
                                           .parameterValues(parameterValues)
                                           .rowMapper(customerCreditMapper())
                                           .pageSize(1000)
                                           .build();}
 
@Bean
public SqlPagingQueryProviderFactoryBean queryProvider() {SqlPagingQueryProviderFactoryBean provider = new SqlPagingQueryProviderFactoryBean();
 
        provider.setSelectClause("select id, name, credit");
        provider.setFromClause("from customer");
        provider.setWhereClause("where status=:status");
        provider.setSortKey("id");
 
        return provider;
}

JdbcPagingItemReader 必须指定一个 PagingQueryProvider，负责提供 SQL 查问语句来按分页返回数据。

上面是一个 JdbcCursorItemReader 的例子代码:

 private JdbcCursorItemReader<Map<String, Object>> buildItemReader(final DataSource dataSource, String tableName,
            String tenant) {JdbcCursorItemReader<Map<String, Object>> itemReader = new JdbcCursorItemReader<>();
        itemReader.setDataSource(dataSource);
        itemReader.setSql("sql here");
        itemReader.setRowMapper(new RowMapper());
        return itemReader;
    }

什么是 Item Writer

既然 ItemReader 是读数据的一个形象，那么 ItemWriter 天然就是一个写数据的形象，它是为每一个 step 提供数据写出的性能。写的单位是能够配置的，咱们能够一次写一条数据，也能够一次写一个 chunk 的数据，对于 chunk 下文会有专门的介绍。ItemWriter 对于读入的数据是不能做任何操作的。

Spring Batch 为 ItemWriter 也提供了十分多的有用的实现类，当然咱们也能够去实现本人的 writer 性能。

什么是 Item Processor

ItemProcessor 对我的项目的业务逻辑解决的一个形象, 当 ItemReader 读取到一条记录之后，ItemWriter 还未写入这条记录之前，I 咱们能够借助 temProcessor 提供一个解决业务逻辑的性能，并对数据进行相应操作。如果咱们在 ItemProcessor 发现一条数据不应该被写入，能够通过返回 null 来示意。ItemProcessor 和 ItemReader 以及 ItemWriter 能够十分好的联合在一起工作，他们之间的数据传输也十分不便。咱们间接应用即可。

chunk 解决流程

spring batch 提供了让咱们依照 chunk 解决数据的能力，一个 chunk 的示意图如下：

它的意思就和图示的一样，因为咱们一次 batch 的工作可能会有很多的数据读写操作，因而一条一条的解决并向数据库提交的话效率不会很高，因而 spring batch 提供了 chunk 这个概念，咱们能够设定一个 chunk size，spring batch 将一条一条解决数据，但不提交到数据库，只有当解决的数据数量达到 chunk size 设定的值得时候，才一起去 commit.

java 的实例定义代码如下：

在下面这个 step 外面，chunk size 被设为了 10，当 ItemReader 读的数据数量达到 10 的时候，这一批次的数据就一起被传到 itemWriter，同时 transaction 被提交。

skip 策略和失败解决

一个 batch 的 job 的 step，可能会解决十分大数量的数据，难免会遇到出错的状况，出错的状况虽呈现的概率较小，然而咱们不得不思考这些状况，因为咱们做数据迁徙最重要的是要保证数据的最终一致性。spring batch 当然也思考到了这种状况，并且为咱们提供了相干的技术支持，请看如下 bean 的配置：

咱们须要注意这三个办法，别离是 skipLimit(),skip(),noSkip(),

skipLimit 办法的意思是咱们能够设定一个咱们容许的这个 step 能够跳过的异样数量，如果咱们设定为 10，则当这个 step 运行时，只有呈现的异样数目不超过 10，整个 step 都不会 fail。留神，若不设定 skipLimit，则其默认值是 0.

skip 办法咱们能够指定咱们能够跳过的异样，因为有些异样的呈现，咱们是能够疏忽的。

noSkip 办法的意思则是指呈现这个异样咱们不想跳过，也就是从 skip 的所以 exception 当中排除这个 exception，从下面的例子来说，也就是跳过所有除 FileNotFoundException 的 exception。

那么对于这个 step 来说，FileNotFoundException 就是一个 fatal 的 exception，抛出这个 exception 的时候 step 就会间接 fail

批处理操作指南

本局部是一些应用 spring batch 时的值得注意的点

批处理准则

在构建批处理解决方案时，应思考以下要害准则和注意事项。

• 批处理体系结构通常会影响体系结构
• 尽可能简化并防止在单批应用程序中构建简单的逻辑构造
• 保持数据的解决和存储在物理上靠得很近（换句话说，将数据保留在处理过程中）。
• 最大限度地缩小系统资源的应用，尤其是 I / O. 在 internal memory 中执行尽可能多的操作。
• 查看应用程序 I / O（剖析 SQL 语句）以确保防止不必要的物理 I / O. 特地是，须要寻找以下四个常见缺点：
• • 当数据能够被读取一次并缓存或保留在工作存储中时，读取每个事务的数据。
  • 从新读取先前在同一事务中读取数据的事务的数据。
  • 导致不必要的表或索引扫描。
  • 未在 SQL 语句的 WHERE 子句中指定键值。
• 在批处理运行中不要做两次一样的事件。例如，如果须要数据汇总以用于报告目标，则应该（如果可能）在最后解决数据时递增存储的总计，因而您的报告应用程序不用重新处理雷同的数据。
• 在批处理应用程序开始时调配足够的内存，以防止在此过程中进行耗时的重新分配。
• 总是假如数据完整性最差。插入适当的检查和记录验证以保护数据完整性。
• 尽可能施行校验和以进行外部验证。例如，对于一个文件里的数据应该有一个数据条数纪录，通知文件中的记录总数以及关键字段的汇总。
• 在具备实在数据量的相似生产环境中尽早打算和执行压力测试。
• 在大批量零碎中，数据备份可能具备挑战性，特地是如果零碎以 24- 7 在线的状况运行。数据库备份通常在在线设计中失去很好的解决，但文件备份应该被视为同样重要。如果零碎依赖于文件，则文件备份过程不仅应该到位并记录在案，还应定期进行测试。

如何默认不启动 job

在应用 java config 应用 spring batch 的 job 时，如果不做任何配置，我的项目在启动时就会默认去跑咱们定义好的批处理 job。那么如何让我的项目在启动时不主动去跑 job 呢？

spring batch 的 job 会在我的项目启动时主动 run，如果咱们不想让他在启动时 run 的话，能够在 application.properties 中增加如下属性：

spring.batch.job.enabled=false

在读数据时内存不够

在应用 spring batch 做数据迁徙时，发现在 job 启动后，执行到肯定工夫点时就卡在一个中央不动了，且 log 也不再打印，期待一段时间之后，失去如下谬误：

红字的信息为：Resource exhaustion event：the JVM was unable to allocate memory from the heap.

翻译过去的意思就是我的项目收回了一个资源耗尽的事件，通知咱们 java 虚拟机无奈再为堆分配内存。

造成这个谬误的起因是: 这个我的项目里的 batch job 的 reader 是一次性拿回了数据库里的所有数据，并没有进行分页，当这个数据量太大时，就会导致内存不够用。解决的方法有两个:

• 调整 reader 读数据逻辑，按分页读取，但实现上会麻烦一些，且运行效率会降落
• 增大 service 内存

起源：blog.csdn.net/topdeveloperr/article/details/84337956

近期热文举荐：

1.1,000+ 道 Java 面试题及答案整顿 (2021 最新版)

2. 终于靠开源我的项目弄到 IntelliJ IDEA 激活码了，真香！

3. 阿里 Mock 工具正式开源，干掉市面上所有 Mock 工具！

4.Spring Cloud 2020.0.0 正式公布，全新颠覆性版本！

5.《Java 开发手册（嵩山版）》最新公布，速速下载！

感觉不错，别忘了顺手点赞 + 转发哦！