如何确保 MultiResourceItemReader 在每次执行作业时刷新资源

Question

我有一个使用 spring 任务调度和批处理作业的 java 应用程序。我依靠job 中的MultiResourceItemReader 从目录中读取一些文件，处理它们，然后删除这些文件。外部进程负责定期将新文件放入该目录中。但问题是，每次job 运行时，它都会尝试读取启动应用程序时存在的相同文件资源，因此失败，因为这些资源已经消失而新文件现在在那里。

问题是，我如何配置应用程序，以便为给定job 的每次计划执行评估资源属性。

相关bean粘贴如下：

  <bean id="multiResourceReader" class="org.springframework.batch.item.file.MultiResourceItemReader">
    <property name="resources" value="file:/opt/data/*.csv" />
    <property name="delegate" ref="testFlatFileItemReader" />
  </bean>

  <batch:job id="MyJob">
    <batch:step id="readandstore">
      <batch:tasklet>
        <batch:chunk reader="multiResourceReader" writer="oracleItemWriter" commit-interval="10" />
      </batch:tasklet>
    </batch:step>
  </batch:job>

  <bean id="runScheduler" class="com.myapp.Scheduler">
    <property name="jobLauncher" ref="jobLauncher" />
    <property name="job" ref="MyJob" />
  </bean>

  <task:scheduled-tasks>
    <task:scheduled ref="runScheduler" method="run" cron="*/30 * * * * *" />
  </task:scheduled-tasks>

Answer 1

在近一周没有收到回复后，我设法解决了我自己发布的问题。

我删除了multiResourceReader 的资源属性，并在实现StepExecutionListener 的<batch:tasklet> 下的侦听器中添加了StepListener。此侦听器有两种方法，一种在步骤执行之前调用，另一种在步骤执行之后调用。 StepListener 接受一个名为 filePattern 的属性，其值将类似于您之前为 multiResourceReader 的资源属性提供的值。以下是更新后的 bean 的样子：

  <batch:job id="MyJob">
    <batch:step id="readandstore">
      <batch:tasklet>
        <batch:chunk reader="multiResourceReader" writer="oracleItemWriter" commit-interval="10" />
        <batch:listeners>
          <batch:listener ref="StepListener" />
        </batch:listeners>
      </batch:tasklet>
    </batch:step>
  </batch:job>

  <bean id="multiResourceReader" class="org.springframework.batch.item.file.MultiResourceItemReader">
    <property name="delegate" ref="csvFileItemReader" />
  </bean>

  <bean id="StepListener" class="com.myapp.StepListener">
    <property name="filePattern" value="file:/opt/data/*.csv" />
  </bean>

这是 com.myapp.StepListener 的样子：

package com.myapp;

import java.io.File;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.apache.commons.io.FileUtils;
import org.springframework.batch.item.file.MultiResourceItemReader;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.io.Resource;

public class StepListener implements StepExecutionListener, ApplicationContextAware {

  private static final Logger logger = Logger.getLogger(StepListener.class.getName());
  private Resource[] resources;
  private ApplicationContext applicationContext;
  private String filePattern;

  @Override
  public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    this.applicationContext = applicationContext;
  }

  public void setFilePattern(String filePattern) {
    this.filePattern = filePattern;
  }

  @Override
  public void beforeStep(StepExecution stepExecution) {

    MultiResourceItemReader reader = (MultiResourceItemReader) applicationContext.getBean("multiResourceReader");
    try {
      resources = applicationContext.getResources(filePattern);
      reader.setResources(resources);
    } catch (IOException ex) {
      logger.log(Level.SEVERE, "Unable to set file resources to bean multiResourceItemReader", ex);
    }
  }

  @Override
  public ExitStatus afterStep(StepExecution stepExecution) {

    if (stepExecution.getExitStatus().equals(ExitStatus.COMPLETED)
            && stepExecution.getStatus().equals(BatchStatus.COMPLETED)
            && resources.length > 0) {

      for (Resource resource : resources) {
        try {
          File oldFile = new File(resource.getFile().getAbsolutePath());
          File newFile = new File(resource.getFile().getAbsolutePath() + ".processed");
          FileUtils.copyFile(oldFile, newFile);
          oldFile.delete();
        } catch (IOException ex) {
          logger.log(Level.SEVERE, "Encountered problem when trying to remove the processed file(s)", ex);
        }
      }
    }

    return stepExecution.getExitStatus();
  }
}

Answer 2

bean multiResourceReader 的作用域是单例的，所以每个只会创建 1 个。当该 bean 被创建时，资源的价值就被解决了。

如果您将 bean multiResourceReader 的范围设置为 step 它也应该可以解决您的问题。然后每次执行该步骤时都会创建一个新的bean。

Answer 3

尝试在名为“multiResourceReader”的配置bean中添加标签“step”：

Answer 4

另一种选择是扩展MultiResourceItemReader 并覆盖open() 方法并在调用super.open() 方法之前调用setResources()。

public class CustomMultiItemReader<T> extends MultiResourceItemReader<T> {
    private static final Log logger = LogFactory.getLog(PrestoMultiItemReader.class);
    
    private String filePattern;

    public CustomMultiItemReader() {
        super();
    }

    @Override
    public void open(ExecutionContext executionContext) {
        Assert.notNull(filePattern, "filePattern must be set");
        try {
            ResourcePatternResolver patternResolver = new PathMatchingResourcePatternResolver();
            Resource[] resources = patternResolver.getResources(filePattern);
            super.setResources(resources);
        } catch (IOException e) {
            logger.error("No resources to read.", e);
        }
        super.open(executionContext);
    }

    public void setFilePattern(String filePattern) {
        this.filePattern = filePattern;
    }
}

您可以拨打以下电话CustomMultiItemReader。您无需致电setResources()。

public CustomMultiItemReader<T> customMultiItemReader() {
    String filePattern = "file:/...";
    CustomMultiItemReader<T> reader = new CustomMultiItemReader<T>();
    reader.setDelegate(reader()); //reader to delegate here
    reader.setFilePattern(filePattern);
    reader.setStrict(true); 
    return reader;
} 

Answer 5

我遇到了同样的问题，但我用另一种方法解决了，也许会有所帮助，我正在使用 FlatFileItemReader，但问题是一样的。

解决方案是在属性资源中使用一个 ref 和一个处理 jobParams 的 bean，这样每次步骤重新开始它都是一个具有新值的新实例，下面的代码就是我的做法：

<bean id="customFileItemReader" class="org.springframework.batch.item.file.FlatFileItemReader" scope="step">
<property name="resource" ref="customFile"/><!-- avoid use directly jobParams here -->
<property name="linesToSkip" value="0" />
<property name="encoding" value="ISO-8859-1" />
<property name="comments" value="*, \u001A" />
<property name="lineMapper">
    <bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
        <property name="lineTokenizer">
            <bean class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
                <property name="delimiter" value=";" />
            </bean>
        </property>
        <property name="fieldSetMapper">
            <bean class="br.com.sample.batch.CustomFieldSetMapperSIAW"/>
        </property>
    </bean>
</property>

<bean id="customFile" class="org.springframework.core.io.FileSystemResource" scope="step">
<constructor-arg value="#{jobParameters[arquivoGSIN]}"></constructor-arg>