spring batch默认的MultiResourceItemReader存在一个严重的bug,应用启动后往读取文件的目录添加文件是无法被读取的,以下代码修复了这个bug
package com.ternnetwork.batch.item.file; import java.io.IOException; import java.util.Arrays; import java.util.Comparator; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ParseException; import org.springframework.batch.item.ResourceAware; import org.springframework.batch.item.UnexpectedInputException; import org.springframework.batch.item.file.ResourceAwareItemReaderItemStream; import org.springframework.batch.item.support.AbstractItemStreamItemReader; import org.springframework.core.io.Resource; import org.springframework.core.io.support.PathMatchingResourcePatternResolver; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; /** * Reads items from multiple resources sequentially - resource list is given by {@link #setResources(Resource[])}, the * actual reading is delegated to {@link #setDelegate(ResourceAwareItemReaderItemStream)}. * * Input resources are ordered using {@link #setComparator(Comparator)} to make sure resource ordering is preserved * between job runs in restart scenario. * * 该类扩展了原来的功能,实现动态获取资源文件列表 * 之前的resources无法监控到文件的增减,此类修复了这个bug * * @author Robert Kasanicky * @author Lucas Ward */ public class ExtendedMultiResourceItemReader<T> extends AbstractItemStreamItemReader<T> { private static final Log logger = LogFactory.getLog(ExtendedMultiResourceItemReader.class); private static final String RESOURCE_KEY = "resourceIndex"; private ResourceAwareItemReaderItemStream<? extends T> delegate; private Resource[] resources; private boolean saveState = true; private int currentResource = -1; // signals there are no resources to read -> just return null on first read private boolean noInput; private boolean strict = false; private String resourcesLocationPattern;//资源路径通配符 /** * In strict mode the reader will throw an exception on * {@link #open(org.springframework.batch.item.ExecutionContext)} if there are no resources to read. * @param strict false by default */ public void setStrict(boolean strict) { this.strict = strict; } private Comparator<Resource> comparator = new Comparator<Resource>() { /** * Compares resource filenames. */ public int compare(Resource r1, Resource r2) { return r1.getFilename().compareTo(r2.getFilename()); } }; public ExtendedMultiResourceItemReader() { this.setExecutionContextName(ClassUtils.getShortName(ExtendedMultiResourceItemReader.class)); } /** * Reads the next item, jumping to next resource if necessary. */ public T read() throws Exception, UnexpectedInputException, ParseException { if (noInput) { return null; } // If there is no resource, then this is the first item, set the current // resource to 0 and open the first delegate. if (currentResource == -1) { currentResource = 0; delegate.setResource(resources[currentResource]); delegate.open(new ExecutionContext()); } return readNextItem(); } /** * Use the delegate to read the next item, jump to next resource if current one is exhausted. Items are appended to * the buffer. * * @return next item from input */ private T readNextItem() throws Exception { T item = readFromDelegate(); while (item == null) { currentResource++; if (currentResource >= resources.length) { return null; } delegate.close(); delegate.setResource(resources[currentResource]); delegate.open(new ExecutionContext()); item = readFromDelegate(); } return item; } private T readFromDelegate() throws Exception { T item = delegate.read(); if(item instanceof ResourceAware){ ((ResourceAware) item).setResource(getCurrentResource()); } return item; } /** * Close the {@link #setDelegate(ResourceAwareItemReaderItemStream)} reader and reset instance variable values. */ public void close() throws ItemStreamException { super.close(); if(!this.noInput) { delegate.close(); } noInput = false; } /** * Figure out which resource to start with in case of restart, open the delegate and restore delegate's position in * the resource. */ public void open(ExecutionContext executionContext) throws ItemStreamException { super.open(executionContext); Assert.notNull(resourcesLocationPattern, "resourcesLocationPattern must be set"); PathMatchingResourcePatternResolver patternResolver=new PathMatchingResourcePatternResolver(); try { resources=patternResolver.getResources(resourcesLocationPattern);//动态读取资源列表 } catch (IOException e) { e.printStackTrace(); } noInput = false; if (resources.length == 0) { if (strict) { throw new IllegalStateException( "No resources to read. Set strict=false if this is not an error condition."); } else { logger.warn("No resources to read. Set strict=true if this should be an error condition."); noInput = true; return; } } Arrays.sort(resources, comparator); if (executionContext.containsKey(getExecutionContextKey(RESOURCE_KEY))) { currentResource = executionContext.getInt(getExecutionContextKey(RESOURCE_KEY)); // context could have been saved before reading anything if (currentResource == -1) { currentResource = 0; } delegate.setResource(resources[currentResource]); delegate.open(executionContext); } else { currentResource = -1; } } /** * Store the current resource index and position in the resource. */ public void update(ExecutionContext executionContext) throws ItemStreamException { super.update(executionContext); if (saveState) { executionContext.putInt(getExecutionContextKey(RESOURCE_KEY), currentResource); delegate.update(executionContext); } } /** * @param delegate reads items from single {@link Resource}. */ public void setDelegate(ResourceAwareItemReaderItemStream<? extends T> delegate) { this.delegate = delegate; } /** * Set the boolean indicating whether or not state should be saved in the provided {@link ExecutionContext} during * the {@link ItemStream} call to update. * * @param saveState */ public void setSaveState(boolean saveState) { this.saveState = saveState; } /** * @param comparator used to order the injected resources, by default compares {@link Resource#getFilename()} * values. */ public void setComparator(Comparator<Resource> comparator) { this.comparator = comparator; } /** * @param resources input resources */ public void setResources(Resource[] resources) { Assert.notNull(resources, "The resources must not be null"); this.resources = Arrays.asList(resources).toArray(new Resource[resources.length]); } public Resource getCurrentResource() { if (currentResource >= resources.length || currentResource < 0) { return null; } return resources[currentResource]; } public String getResourcesLocationPattern() { return resourcesLocationPattern; } public void setResourcesLocationPattern(String resourcesLocationPattern) { this.resourcesLocationPattern = resourcesLocationPattern; } }
关键代码
resources=patternResolver.getResources(resourcesLocationPattern);//动态读取资源列表
配置示例
<bean id="bpMultiResourceReader" scope="prototype" class="com.ternnetwork.batch.item.file.ExtendedMultiResourceItemReader"> <property name="resourcesLocationPattern" value="file:${batch.input.dir}*.csv"/> <property name="delegate" ref="bpFlatFileItemReader"/> </bean>
如果您觉得我的文章给了您帮助,请为我买一杯饮料吧!我将非常感激并坚持为大家提供更多帮助!
相关推荐
主要给大家介绍了Spring Batch读取txt文件并写入数据库的方法,SpringBatch 是一个轻量级、全面的批处理框架。这里我们用它来实现文件的读取并将读取的结果作处理,处理之后再写入数据库中的功能。需要的朋友可以...
本篇文章主要介绍了spring batch 读取多个文件数据导入数据库示例,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
Spring Batch是Spring的一个子项目,使用Java语言并基于Spring框架为基础开发,使得已经使用 Spring 框架的开发者或者企业更容易访问和利用企业服务。 Spring Batch 提供了大量可重用的组件,包括了日志、追踪、事务、...
Spring Batch API(Spring Batch 开发文档).CHM。 官网 Spring Batch API,Spring Batch 开发文档
Spring Boot整合Spring Batch的一个小例子,在网上发现这方面的资源比较少,特此将其上传供大家学习。
spring-batch4.0.0 batch spring-batch集成 spring-batch.jar
Spring Batch是一个轻量级的,完全面向Spring的批处理框架,可以应用于企业级大量的数据处理系统。Spring Batch以POJO和大家熟知的Spring框架为基础,使开发者更容易的访问和利用企业级服务。Spring Batch可以提供...
基于Spring Batch的大数据量并行处理 基于Spring Batch的大数据量并行处理
最近在研究springBoot+springbatch ,按照官网的实例做了一个实例。 最近在研究springBoot+springbatch ,按照官网的实例做了一个实例。
spring batch批处理框架和对应的源码资源 rar 可以直接运行的
Work with all aspects of batch processing in a modern Java environment using a selection of Spring frameworks. This book provides up-to-date examples using the latest configuration techniques based on...
Spring Batch是一个基于Spring的企业级批处理框架,所谓企业批处理就是指在企业级应用中,不需要人工干预,定期读取数据,进行相应的业务处理之后,再进行归档的这类操作。SpringBatch为我们提供了统一的读写接口、...
Spring Batch in Action is a comprehensive, in-depth guide to writing batch applications using Spring Batch. Written for developers who have basic knowledge of Java and the Spring lightweight ...
NULL 博文链接:https://virusfu.iteye.com/blog/1150730
2,什么是 Spring Batch 3 二,Spring Batch结构 4 1,Spring Batch体系结构 4 2,Spring Batch主要对象 5 三,Spring Batch流程介绍 5 四,Spring Batch之Step执行过程介绍 6 五,Spring Batch应用 7 1,简单应用 7...
mybatis、springBatch、mysql、quartz、spring、springMVC 部署说明: 本项目为两个数据库,由一个数据库的表向另外一个数据库的表做数据迁移,其中数据库脚本在:/src/main/resources/sql/下面(其中data_rep中的表...
基本篇重点讲述了数据批处理的核心概念、典型的作业配置、作业步配置,以及Spring Batch框架中经典的三步走策略:数据读、数据处理和数据写,详尽地介绍了如何对CVS格式文件、JSON格式文件、XML文件、数据库和JMS...
难得的详细spring batch资料 难得的详细spring batch资料
SpringBatch数据库建表语句,存储springBatch批处理过程中需要保存的数据和步骤信息
spring-batchr的分区示例的简单实现。