FlatFileItemWriter 应该写入与输入文件同名的输出文件

Question

我有一个 spring 批处理作业，它从目录中读取匹配命名模式的文件，进行一些处理，然后在输入文件中写回每行的处理状态。编写器必须生成与输入文件同名的输出文件。我将模式传递给 MultiResourceItemReader：“files-*.txt”，并期望 FlatFileItemWriter 使用输入文件的名称。如何在上下文 xml 文件中指定此约束？

阅读器 bean

<bean id="multiResourceReader" class="org.springframework.batch.item.file.MultiResourceItemReader" scope="step">
  <property name="resources" value="file:#{jobParameters['cwd']}/#{jobParameters['inputFolder']}/file-*.txt" />
  <property name="delegate" ref="itemReader" />
</bean>
<bean id="itemReader" class="org.springframework.batch.item.file.FlatFileItemReader" scope="step">

        <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="," />
                <property name="names" value="paramA,paramB" />
              </bean>
            </property>
            <property name="fieldSetMapper" >
              <bean class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper">
                <property name="targetType" value="a.b.c.d.MyObject" />
              </bean>
            </property>
          </bean>
        </property>
    </bean>

我试过这个：在我的 FlatFileItemReader 实现中，将输入文件路径作为字符串属性添加到 ExecutionContext 的映射中。在我的 FlatFileItemWriter 实现中 - 覆盖 setResource 并根据 ExecutionContext 的值实际创建一个 Resource 对象。这可以吗？

Answer 1

我认为这个解决方案可能会产生错误的输出，因为MultiResourceItemReader 从第一个文件中读取了一堆项目，如果这个集合与您的完成策略不匹配，它会从下一个资源中读取；在这种情况下，您将拥有来自两个不同来源的项目，并且itemReader bean 未同步以写入正确的输出资源。
您可以尝试解决 let a.b.c.d.MyObject 实现 org.springframework.batch.item.ResourceAware 并编写自定义多资源项编写器并在资源更改时管理资源打开/关闭（此解决方案并非微不足道，因为需要状态保存才能重新启动，当然）。

class MyItemWriter implements ItemWriter<MyObject>,ItemStream {
  private FlatFileItemWriter delegate;
  private String resourceName;

  public open(ExecutionContext e) {
    resourceName = e.getString("resourceName");
    delegate.open(e);
  }
  public close(ExecutionContext e) {
    e.putString("resourceName",resourceName);
    delegate.close(e);
  }
  public update(ExecutionContext e) {
    e.putString("resourceName",resourceName);
    delegate.update(e);
  }
  public void write(List<? extends MyObject> items) {
    for(final MyObject o : items) {
      String itemResource = o.getResource().toString();
      if(!itemResource.equals(resourceName)) {
        closeDelegate();
        openDelegate();
      }
      writeToDelegate(o);
    }
  }
}

当然，这没有经过测试，只是一个想法。
另一种方式完全不同，并以这种方式使用decider 模拟循环：

1. 创建一个bean（例如称为ResourceHolder）用于存储来自file:#{jobParameters['cwd']}/#{jobParameters['inputFolder']}/file-*.txt的资源；这个 bean 没有存储到执行上下文中，而是在每次作业开始时填充一个 JobExecutionListener，因为执行上下文应该保持很小。
2. 使用判定器模拟循环！从决策者将当前资源的名称存储到执行上下文中；您还需要存储当前资源的索引以移动到下一个或停止您的工作
3. 读取/处理/写入步骤使用一个文件，该文件从决策程序存储到执行上下文中

---

<batch:job id="job">
  <batch:decision decider="decider" id="decider">
    <batch:next on="CONTINUABLE" to="readWriteStep" />
    <batch:end on="FINISHED" />
  </batch:decision>
  <batch:step id="readWriteStep" reader="itemReader" write="itemWriter" next="decider" />
  <batch:listeners>
    <batch:listener class="ResourceHolderFiller">
      <property name="resources" value="file:#{jobParameters['cwd']}/#{jobParameters['inputFolder']}/file-*.txt" />
      <property name="resourceHolderBean" ref="resourceHolderBean" />
    </batch:listener>
  </batch:listeners>
</batch:job>
<bean id="resourceHolder" class="ResourceHolderBean" />
<bean id="decider" class="MyDecider">
  <property name="resourceHolderBean" ref="resourceHolderBean" />
</bean>

class ResourceHolderBean
{
  Resource[] resource;

  public void setResource(Resource[] resource) {...}
  public Resource[] getResource() {...}
}
class MyDecider implements JobExecutionDecider {
  ResourceHolderBean resourceBean;

  public void setResourceBean(ResourceHolderBean resBean){...}

  FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
    Integer resourceIndex = jobExecution.getExecutionContext().getInt("resourceIndex");
    if(resourceIndex == null)
    {
      resourceIndex = 0;
    }
    if(resourceIndex == resourceBean.getResources().size()) {
      return new FlowExecutionStatus(RepeatStatus.FINISHED.name());
    }
    else {
      jobExecution.getExecutionContext().putInt("resourceIndex",resourceIndex);
      // This is the key your reader/writer must use as inpu/output filename
      jobExecution.getExecutionContext().putString("resourceName",resourceBean.getResources()[resourceIndex].toString());
    }
  }
}

ResourceHolderFiller 是一个将资源注入 ResourceHolderBean bean 的作业侦听器。

希望能有所帮助。