Spring Batch -批处理的域语言

对于任何有经验的批处理架构师，批处理的总体概念用于 春季批次应该是熟悉和舒适的。有“作业”和“步骤”和 开发人员提供的处理单元称为 和 。然而 由于 Spring 模式、操作、模板、回调和习语，有 以下机会：​​ItemReader​​​​ItemWriter​​

• 在遵守明确关注点分离方面有了显著改善。
• 清晰描述的架构层和作为接口提供的服务。
• 简单和默认的实现，允许快速采用和易于使用 开箱即用。
• 显著增强了可扩展性。

下图是批处理参考体系结构的简化版本，该体系结构 已经使用了几十年。它概述了构成 批处理的域语言。此体系结构框架是一个蓝图，具有 通过过去几代 平台（大型机上的COBOL，Unix上的C，现在任何地方的Java）。JCL 和 COBOL 开发人员 他们可能与 C、C# 和 Java 开发人员一样熟悉这些概念。春天 批处理提供层、组件和技术的物理实现 在用于解决 使用基础架构和扩展创建从简单到复杂的批处理应用程序 满足非常复杂的加工需求。

图1.批量构造型

上图突出显示了构成域语言的关键概念 春季批次。A 有一到多个步骤，每个步骤正好有一个， 一，一.需要启动作业（使用 ），并且需要存储有关当前正在运行的进程的元数据（在 中）。​​Job​​​​ItemReader​​​​ItemProcessor​​​​ItemWriter​​​​JobLauncher​​​​JobRepository​​

工作

本节介绍与批处理作业概念相关的构造型。A 是 封装整个批处理过程的实体。与其他弹簧一样常见 项目，a 与 XML 配置文件或基于 Java 的连接在一起 配置。此配置可能称为“作业配置”。但是，它只是整个层次结构的顶部，如下图所示：​​Job​​​​Job​​​​Job​​

图2.作业层次结构

在Spring Batch中，a只是实例的容器。它结合了多个 逻辑上属于流中并允许配置属性的步骤 全局到所有步骤，例如可重启性。作业配置包含：​​Job​​​​Step​​

• 作业的名称。
• 实例的定义和排序。Step
• 作业是否可重新启动。

作业实例

A 是指逻辑作业运行的概念。考虑一个批处理作业 应该在一天结束时运行一次，例如从前面 图。有一个工作，但每个单独的运行都必须 单独跟踪。对于此作业，每天有一个逻辑。 例如，有 1 月 1 日运行、1 月 2 日运行等。如果 1 月 1 日 运行第一次失败，第二天再次运行，它仍然是 1 月 1 日的运行。 （通常，这也对应于它正在处理的数据，这意味着一月 第一次运行处理 1 月 1 日的数据）。因此，每个都可以有多个 执行（本章后面将更详细地讨论），并且仅 一个（对应于特定的和识别的）可以 在给定时间运行。​​JobInstance​​​​EndOfDay​​​​Job​​​​EndOfDay​​​​Job​​​​JobInstance​​​​JobInstance​​​​JobExecution​​​​JobInstance​​​​Job​​​​JobParameters​​

a 的定义与要加载的数据完全无关。 完全由实现来确定如何加载数据。为 例如，在方案中，数据上可能有一列指示数据所属的 OR。所以，1月1日运行 将仅加载来自 1 次的数据，而 1 月 2 日运行将仅使用来自 2nd. 由于此决定很可能是商业决策，因此由 THE 决定。但是，使用相同的决定了是否 “状态”（即本章后面讨论的 ） 使用以前的执行。使用新手段“从 开始“，使用现有实例通常意味着”从你离开的地方开始 关”。​​JobInstance​​​​ItemReader​​​​EndOfDay​​​​effective date​​​​schedule date​​​​ItemReader​​​​JobInstance​​​​ExecutionContext​​​​JobInstance​​

作业参数

讨论过以及它与 有何不同，自然要问的问题 是：“一个如何与另一个区别？答案是：。对象包含一组用于启动批处理的参数 工作。它们可用于识别，甚至可以用作运行期间的参考数据，如 下图显示：​​JobInstance​​​​Job​​​​JobInstance​​​​JobParameters​​​​JobParameters​​

图3.作业参数

在前面的示例中，有两个实例，一个用于 1 月 1 日，另一个用于 1 月 1 日 对于 1 月 2 日，实际上只有一个 ，但它有两个对象： 一个以作业参数 01-01-2017 启动，另一个以 参数 01-02-2017。因此，合约可以定义为：= + 标识。这允许开发人员有效地控制如何定义 a，因为他们控制传入的参数。​​Job​​​​JobParameter​​​​JobInstance​​​​Job​​​​JobParameters​​​​JobInstance​​

作业执行

A 是指单次尝试运行作业的技术概念。一 执行可能以失败或成功告终，但对应于给定的 除非执行成功完成，否则不会认为执行已完成。 以前面描述的为例，考虑一个 01-01-2017 第一次运行时失败。如果再次以相同的方式运行 将作业参数标识为首次运行 （01-01-2017），新的 创建。但是，仍然只有一个.​​JobExecution​​​​JobInstance​​​​EndOfDay​​​​Job​​​​JobInstance​​​​JobExecution​​​​JobInstance​​

A 定义什么是作业以及如何执行作业，A 是 纯粹的组织对象将执行组合在一起，主要是为了能够正确 重新启动语义。然而，A 是 在运行期间实际发生，并且包含更多必须控制的属性 并持久化，如下表所示：​​Job​​​​JobInstance​​​​JobExecution​​

表 1.作业执行属性

这些属性很重要，因为它们是持久化的，可用于完全 确定执行的状态。例如，如果 01-01 的作业是 在晚上 9：00 执行，在 9：30 失败，批处理中将进行以下条目 元数据表：​​EndOfDay​​

表 2.BATCH_JOB_INSTANCE

表 3.BATCH_JOB_EXECUTION_PARAMS

表 4.BATCH_JOB_EXECUTION

现在作业已失败，假设问题花了整整一夜的时间 已确定，以便“批处理窗口”现在关闭。进一步假设窗口 从晚上 9：00 开始，工作在 01-01 再次启动，从中断的地方开始，然后 9：30成功完成。因为现在是第二天，所以01-02工作必须是 也运行，之后在 9：31 开始，并以正常方式完成 小时时间为10：30。没有要求在之后踢一个 另一个，除非两个作业有可能尝试访问相同的数据， 导致数据库级别的锁定问题。这完全取决于调度程序 确定何时应运行 。由于它们是分开的，春天 批处理不会尝试阻止它们并发运行。（正在尝试运行 同样，而另一个已经在运行导致被抛出）。现在应该有一个额外的条目 在 和 表中以及表中的两个额外条目中，如下表所示：​​JobInstance​​​​Job​​​​JobInstances​​​​JobInstance​​​​JobExecutionAlreadyRunningException​​​​JobInstance​​​​JobParameters​​​​JobExecution​​

表 5.BATCH_JOB_INSTANCE

表 6.BATCH_JOB_EXECUTION_PARAMS

表 7.BATCH_JOB_EXECUTION

步

A 是一个域对象，它封装批处理的独立顺序阶段 工作。因此，每个步骤完全由一个或多个步骤组成。A 包含 定义和控制实际批处理所需的所有信息。这 是一个必然模糊的描述，因为任何给定的内容都在 开发人员编写 .A 可以像 开发人员的愿望。一个简单的可以将数据从文件加载到数据库中， 需要很少或不需要代码（取决于所使用的实现）。更复杂的可能具有作为处理的一部分应用的复杂业务规则。如 对于 ，a 具有与唯一 相关联的个体，如下图所示：​​Step​​​​Job​​​​Step​​​​Step​​​​Job​​​​Step​​​​Step​​​​Step​​​​Job​​​​Step​​​​StepExecution​​​​JobExecution​​

图4.包含步骤的作业层次结构

步骤执行

A 表示执行 的单次尝试。每次运行 时都会创建一个新节点，类似于 。但是，如果步骤失败 执行，因为之前的步骤失败，不会为其保留任何执行。仅当 A 实际启动时才会创建。​​StepExecution​​​​Step​​​​StepExecution​​​​Step​​​​JobExecution​​​​StepExecution​​​​Step​​

​​Step​​执行由类的对象表示。每次执行 包含对其相应步骤和事务相关的引用 数据，例如提交和回滚计数以及开始和结束时间。此外，每一步 执行包含 ，其中包含开发人员需要的任何数据 在批处理运行中保留，例如所需的统计信息或状态信息 重新启动。下表列出了 的属性：​​StepExecution​​​​JobExecution​​​​ExecutionContext​​​​StepExecution​​

表 8.步骤执行属性

执行上下文

A 表示持久化的键/值对的集合，并且 由框架控制，为开发人员提供一个存储持久化的地方 作用域为对象或对象的状态。（对于那些 熟悉石英，它与 .）最好的用法示例是 便于重启。以平面文件输入为例，同时处理单个 行，框架会定期保留 at 提交点。行为 所以让存储其状态，以防在运行期间发生致命错误 或者即使停电了。所需要的只是输入当前的行数 读入上下文，如以下示例所示，框架执行 休息：​​ExecutionContext​​​​StepExecution​​​​JobExecution​​​​JobDataMap​​​​ExecutionContext​​​​ItemReader​​

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

以构造型部分中的示例为例，假设 是将文件加载到数据库中的一个步骤。在第一次失败的运行之后， 元数据表类似于以下示例：​​EndOfDay​​​​Job​​​​loadData​​

表 9.BATCH_JOB_INSTANCE

表 10.BATCH_JOB_EXECUTION_PARAMS

表 11.BATCH_JOB_EXECUTION

表 12.BATCH_STEP_EXECUTION

表 13.BATCH_STEP_EXECUTION_CONTEXT

在前面的案例中，运行了 30 分钟并处理了 40，321 个“件”，其中 将表示在这种情况下文件中的行。此值在每个值之前更新 由框架提交，并且可以包含与 中的条目对应的多行。在提交之前收到通知需要各种实现（或 ）之一，这将更详细地讨论 本指南的后面部分。与前面的示例一样，假定 第二天重新启动。重新启动时，来自 最后一次运行是从数据库重构的。当打开时，它可以 检查它在上下文中是否有任何存储状态并从那里初始化自身， 如以下示例所示：​​Step​​​​ExecutionContext​​​​StepListener​​​​ItemStream​​​​Job​​​​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();
    }
}

在这种情况下，在前面的代码运行后，当前行为 40，322，让从中断的地方重新开始。您也可以使用 for 需要保留的有关运行本身的统计信息。例如，如果平面文件 包含跨多行存在的处理订单，可能需要 存储已处理的订单数量（这与 行读取），以便可以在末尾发送一封电子邮件，其中包含总数 在正文中处理的订单。框架为开发人员处理存储它， 以正确将其范围限定为个人.可能很难 知道是否应该使用现有的。例如，使用上面的示例，当 01-01 运行第二次再次开始时， 框架认识到它是相同的，并且在个人基础上， 从数据库中提取 ，并将其（作为 的一部分）交给它自己。相反，对于 01-02 运行，框架 认识到它是一个不同的实例，因此必须将空上下文交给 .框架为以下类型做出了许多此类决定 开发人员，以确保在正确的时间向他们提供状态。这也很重要 请注意，在任何给定时间都只存在一个。 的客户端应该小心，因为这会创建一个共享的 键空间。因此，在输入值时应小心，以确保没有数据 覆盖。但是，上下文中绝对没有存储数据，因此没有 对框架产生不利影响的方式。​​Step​​​​ExecutionContext​​​​Step​​​​JobInstance​​​​ExecutionContext​​​​EndOfDay​​​​JobInstance​​​​Step​​​​ExecutionContext​​​​StepExecution​​​​Step​​​​Step​​​​ExecutionContext​​​​StepExecution​​​​ExecutionContext​​​​Step​​

请注意，每个至少有一个，每个 一个。例如，请考虑以下事项 代码片段：​​ExecutionContext​​​​JobExecution​​​​StepExecution​​

ExecutionContext ecStep = stepExecution.getExecutionContext();
ExecutionContext ecJob = jobExecution.getExecutionContext();
//ecStep does not equal ecJob

如评论中所述，不等于.它们是两个不同的.作用域为 的 保存在 中的每个提交点，而作用域为 Job 的提交点则保存在每次执行之间。​​ecStep​​​​ecJob​​​​ExecutionContexts​​​​Step​​​​Step​​​​Step​​

作业存储库

​​JobRepository​​是前面提到的所有构造型的持久性机制。 它为 、 和实现提供 CRUD 操作。首次启动 a 时，将从存储库中获取 a。此外，在 执行过程和实现持久化 通过将它们传递到存储库。​​JobLauncher​​​​Job​​​​Step​​​​Job​​​​JobExecution​​​​StepExecution​​​​JobExecution​​

Spring Batch XML 命名空间为配置实例提供支持 ，如以下示例所示：​​JobRepository​​​​<job-repository>​​

<job-repository id="jobRepository"/>

使用 Java 配置时，注释提供自动配置的组件之一。​​@EnableBatchProcessing​​​​JobRepository​​

作业启动器

​​JobLauncher​​表示一个简单的接口，用于启动具有给定集合的 ，如以下示例所示：​​Job​​​​JobParameters​​

public interface JobLauncher {

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

期望实现从 获取有效 并执行 .​​JobExecution​​​​JobRepository​​​​Job​​

项目阅读器

​​ItemReader​​是一个抽象，表示对 、 的输入的检索 一次项目。当用尽它可以提供的项目时，它 通过返回 来指示这一点。您可以找到有关界面及其的更多详细信息 读取器和编写器中的各种实现。​​Step​​​​ItemReader​​​​null​​​​ItemReader​​

条目编写器

​​ItemWriter​​是一个抽象，表示一个、一个批处理或块的输出 一次的项目。通常，不知道它应该输入的输入 接收下一个，并且只知道在其当前调用中传递的项目。您可以找到更多 有关接口及其在读取器和编写器中的各种实现的详细信息。​​Step​​​​ItemWriter​​​​ItemWriter​​

项目处理器

​​ItemProcessor​​是表示项的业务处理的抽象。 当读取一个项目和写入一个项目时，提供了一个访问点来转换或应用其他业务处理。 如果在处理项目时确定该项目无效，则返回指示不应写出该项目。您可以在阅读器和编写器中找到有关界面的更多详细信息。​​ItemReader​​​​ItemWriter​​​​ItemProcessor​​​​null​​​​ItemProcessor​​

批处理命名空间

前面列出的许多域概念都需要在 Spring 中进行配置。虽然有上述接口的实现，但您可以 在标准 Bean 定义中使用，为了便于 配置，如以下示例所示：​​ApplicationContext​​

<beans:beans xmlns="http://www.springframework.org/schema/batch"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
   http://www.springframework.org/schema/beans
   https://www.springframework.org/schema/beans/spring-beans.xsd
   http://www.springframework.org/schema/batch
   https://www.springframework.org/schema/batch/spring-batch.xsd">

<job id="ioSampleJob">
    <step id="step1">
        <tasklet>
            <chunk reader="itemReader" writer="itemWriter" commit-interval="2"/>
        </tasklet>
    </step>
</job>

</beans:beans>

只要声明了批处理命名空间，就可以使用其任何元素。您可以找到更多 有关在配置和配置中配置作业的信息 运行作业。您可以在配置步骤中找到有关配置 的更多信息。​​Step​​