网站建设做软件开发吗,河南省通信管理局网站备案电话,网站移动端建设,淘宝客15套单页网站程序模板打包在之前的几篇文章中#xff0c;我们已经讨论了如何启动和运行Spring Batch。 现在#xff0c;我们将开始讨论可用于扩展Spring Batch的一些策略。 本文将重点介绍如何对步骤进行分区#xff0c;以使该步骤具有多个线程#xff0c;每个线程并行处理一块数据。 如果您有大量… 在之前的几篇文章中我们已经讨论了如何启动和运行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 MapString, ExecutionContext partition(int gridSize) {MapString, ExecutionContext partitionMap new HashMapString, 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 idinventoryFilePartitionerclasscom.keyhole.example.partition.MultiFileResourcePartitionerscopestepproperty nameinboundDir value/data/inbound /
/bean 查看实现的分区方法我们只列出所有在指定入站目录中找到的文件并为找到的每个文件创建一个ExecutionContext并将其添加到返回的地图中。 用于将每个ExecutionContext放入映射中的唯一键也将成为为映射中的每个条目创建的步骤名称的一部分。 Spring Batch将使用该分区映射根据映射中找到的每个键创建一个从属步骤。 要对步骤进行分区您需要首先创建分区配置将引用的步骤。 该步骤应与Spring Batch中的任何其他步骤一样进行配置在此示例中我们将定义FlatFileItemReader和简单的ItemWriter 它们将仅调用toString方法并将其记录到控制台。 这是该踏板及其相关弹簧弹片的配置详细信息。 这里要注意的重要一点是我们将ItemReader的作用域限定在步骤级别这样我们就不会在使用同一bean处理数据的多个线程中遇到任何问题。 我们还需要以这种方式确定它们的作用域以便我们可以使用Spring后期绑定在Step的ExecutionContext中指定保存文件资源的值。 batch:step idinventoryLoadStepxmlnshttp://www.springframework.org/schema/batchbatch:tasklet transaction-managertransactionManagerbatch:chunk readerinventoryLoadReader writerlogItemWritercommit-interval5000 //batch:tasklet
/batch:step
bean nameinventoryLoadReader scopestepclassorg.springframework.batch.item.file.FlatFileItemReaderproperty nameresourcevalue#{stepExecutionContext[fileResource]} /property namelineMapper refinventoryLineMapper /
property namelinesToSkip value1 /
/beanbean nameinventoryLineMapperclassorg.springframework.batch.item.file.mapping.DefaultLineMapperproperty namefieldSetMapper refinventoryFieldMapper /property namelineTokenizer refinventoryLineTokenizer /
/beanbean nameinventoryLineTokenizer classorg.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 FieldSetMapperInventoryItem {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 idInventoryLoaderbatch:step idpartitionedInventoryLoadStepbatch:partition stepinventoryLoadStep partitionerinventoryFilePartitionerbatch:handler grid-size10 task-executorinventoryLoadTaskExecutor //batch:partition/batch:step
/batch:job 在配置分区步骤时您可以像定义其他步骤一样定义一个步骤方法是为其指定一个ID并根据需要指定下一步骤的值。 Spring Batch没有将步骤的内容定义为普通的块或任务集而是提供了一个分区标签该标签要求您指定要分区的作业步骤以及将用于确定数据块的Partitioner。 您还需要定义将要处理这些步骤的分区处理程序在这种情况下我们将使用ThreadPoolTaskExecutor 该线程处理程序的线程池大小为10如果不使用它们则允许它们超时。 bean idinventoryLoadTaskExecutorclassorg.springframework.scheduling.concurrent.ThreadPoolTaskExecutorproperty namecorePoolSize value10 /property namemaxPoolSize value10 /property nameallowCoreThreadTimeOut valuetrue /
/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 –步骤分区 。 翻译自: https://www.javacodegeeks.com/2013/12/scaling-spring-batch-step-partitioning.html
