扩展Spring Batch –步骤分区

在之前的几篇文章中，我们已经讨论了如何启动和运行Spring Batch。现在，我们将开始讨论可用于扩展Spring Batch的一些策略。

本文将重点介绍如何对步骤进行分区，以使该步骤具有多个线程，每个线程并行处理一块数据。如果您有大量的数据可以在逻辑上拆分为可以并行处理的较小的块，这将非常有帮助。这种工作方式是，您将定义一个主要步骤，该步骤负责确定块的基础，然后将所有这些块都植入到一组从属步骤中，以处理每个块。

分区

如果我能回到过去的经验，那么一个很好的例子就是在大型采购系统中处理每个公司的所有每日发票。我们将要处理的数据可以按处理的每个公司在逻辑上进行拆分。假设有250家公司参与此采购系统，我们的分区步骤已定义为具有15个线程。我们的分区程序可能会运行查询，以查找当天有等待处理发票的所有公司。此时，分区程序的职责是为每个公司创建一个ExecutionContext并将其添加到具有唯一键的地图中。该ExecutionContext应该包含处理该公司的发票所需的任何信息，例如公司ID和任何其他相关信息。当分区程序返回ExecutionContexts的映射时，Spring Batch将为映射中的每个条目创建一个新的Step ，并将键值用作步骤名称的一部分。根据配置（例如15个线程），它将创建15个线程的池并开始一次并行执行15个步骤。例如，如果您有85个步骤，Spring Batch将开始执行15个步骤，并且在完成每个步骤后，完成该步骤的线程将接下一个步骤并开始执行，直到所有步骤都已完成。

一个例子

现在，我们对分区的工作原理有了基本的了解，让我们看一个简单的示例。对于我们的用例，我们将检查入站目录，传入的供应商目录文件将被转储到该目录中。因此，要创建一个Spring Batch分区程序，我们需要创建一个实现Spring Batch的Partitioner接口的类。由于这是通用的并且可以重用，所以我们将调用此类MultiFileResourcePartitioner ，这是一个简单的POJO，并且只有一个字段名称“ inboundDir”代表包含要处理文件的目录的路径。 Partitioner接口指定该类应实现一个名为“ partition”的方法，该方法采用一个表示网格大小的int参数，并返回一个保存ExecutionContext的Map。

这是MultiFileResourcePartitioner的类列表：

package com.keyhole.example.partition;import java.io.File;
import java.util.HashMap;
import java.util.Map;import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.core.io.FileSystemResource;public class MultiFileResourcePartitioner implements Partitioner {private String inboundDir;@Overridepublic Map<String, ExecutionContext> partition(int gridSize) {Map<String, ExecutionContext> partitionMap = new HashMap<String, ExecutionContext>();File dir = new File(inboundDir);if (dir.isDirectory()) {File[] files = dir.listFiles();for (File file : files) {ExecutionContext context = new ExecutionContext();context.put("fileResource", file.toURI().toString());partitionMap.put(file.getName(), context);}}return partitionMap;}public String getInboundDir() {return inboundDir;}public void setInboundDir(String inboundDir) {this.inboundDir = inboundDir;}}

在我们的应用程序上下文中，此bean的配置如下所示：

<bean id="inventoryFilePartitioner"class="com.keyhole.example.partition.MultiFileResourcePartitioner"scope="step"><property name="inboundDir" value="/data/inbound" />
</bean>

查看实现的分区方法，我们只列出所有在指定入站目录中找到的文件，并为找到的每个文件创建一个ExecutionContext并将其添加到返回的地图中。用于将每个ExecutionContext放入映射中的唯一键也将成为为映射中的每个条目创建的步骤名称的一部分。 Spring Batch将使用该分区映射根据映射中找到的每个键创建一个从属步骤。

要对步骤进行分区，您需要首先创建分区配置将引用的步骤。该步骤应与Spring Batch中的任何其他步骤一样进行配置，在此示例中，我们将定义FlatFileItemReader和简单的ItemWriter ，它们将仅调用toString（）方法并将其记录到控制台。

这是该踏板及其相关弹簧弹片的配置详细信息。这里要注意的重要一点是，我们将ItemReader的作用域限定在步骤级别，这样我们就不会在使用同一bean处理数据的多个线程中遇到任何问题。我们还需要以这种方式确定它们的作用域，以便我们可以使用Spring后期绑定在Step的ExecutionContext中指定保存文件资源的值。

<batch:step id="inventoryLoadStep"xmlns="http://www.springframework.org/schema/batch"><batch:tasklet transaction-manager="transactionManager"><batch:chunk reader="inventoryLoadReader" writer="logItemWriter"commit-interval="5000" /></batch:tasklet>
</batch:step>
<bean name="inventoryLoadReader" scope="step"class="org.springframework.batch.item.file.FlatFileItemReader"><property name="resource"value="#{stepExecutionContext['fileResource']}" /><property name="lineMapper" ref="inventoryLineMapper" />
<property name="linesToSkip" value="1" />
</bean><bean name="inventoryLineMapper"class="org.springframework.batch.item.file.mapping.DefaultLineMapper"><property name="fieldSetMapper" ref="inventoryFieldMapper" /><property name="lineTokenizer" ref="inventoryLineTokenizer" />
</bean><bean name="inventoryLineTokenizer" 		class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer" />

由于在此示例中我们正在读取和处理以逗号分隔的文本文件，因此对于此步骤配置，我们只需编写很少的代码。我们唯一需要实现的代码是将行的内容映射到表示文件记录的对象所需的FieldSetMapper 。文件中的每一行都将包含“类别”，“子类别”，“描述”，“目录编号”，“颜色”，“尺寸”，“价格”和“数量”字段。因此，我们的对象将包含这些字段，并且我们的FieldSetMapper代码清单将如下所示。

package com.keyhole.example.partition;import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.FieldSet;
import org.springframework.stereotype.Component;
import org.springframework.validation.BindException;@Component("inventoryFieldMapper")
public class InventoryItemFieldSetMapper implements FieldSetMapper<InventoryItem> {@Overridepublic InventoryItem mapFieldSet(FieldSet fieldSet) throws BindException {InventoryItem item = new InventoryItem();item.setCategory(fieldSet.readString(0));item.setSubCategory(fieldSet.readString(1));item.setDescription(fieldSet.readString(2));item.setCatalogNum(fieldSet.readString(3));item.setColor(fieldSet.readString(4));item.setSize(fieldSet.readString(5));item.setPrice(fieldSet.readDouble(6));item.setQty(fieldSet.readInt(7));return item;}}

现在我们已经创建并配置了Partitioner和Step，剩下要做的就是配置分区步骤本身！就像这样简单：

<batch:job id="InventoryLoader"><batch:step id="partitionedInventoryLoadStep"><batch:partition step="inventoryLoadStep" partitioner="inventoryFilePartitioner"><batch:handler grid-size="10" task-executor="inventoryLoadTaskExecutor" /></batch:partition></batch:step>
</batch:job>

在配置分区步骤时，您可以像定义其他步骤一样定义一个步骤，方法是为其指定一个ID，并根据需要指定下一步骤的值。 Spring Batch没有将步骤的内容定义为普通的块或任务集，而是提供了一个分区标签，该标签要求您指定要分区的作业步骤以及将用于确定数据块的Partitioner。您还需要定义将要处理这些步骤的分区处理程序，在这种情况下，我们将使用ThreadPoolTaskExecutor ，该线程处理程序的线程池大小为10，如果不使用它们，则允许它们超时。

<bean id="inventoryLoadTaskExecutor"class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"><property name="corePoolSize" value="10" /><property name="maxPoolSize" value="10" /><property name="allowCoreThreadTimeOut" value="true" />
</bean>

因此，如果您的Spring Batch流程具有处理大量记录的步骤，并且您对提高性能感兴趣，请考虑尝试进行步骤分区。它应该易于实现，并为您提供一些其他性能，以帮助加快处理时间。

其他资源

对于与本文相关的示例代码，我已将源代码上传到位于https://github.com/jonny-hackett/batch-example的github存储库中。要执行与本文相关的示例代码，有一个JUnit测试名称InventoryLoadTest 。数据文件位于src / test / resources / data / inbound下，需要放置在与Partitioner入站目录匹配的本地目录中。另外，请访问http://docs.spring.io/spring-batch/reference/html/scalability.html 。

参考：在Keyhole Software博客上从我们的JCG合作伙伴 Keyhole Software 扩展Spring Batch –步骤分区。