java - 如何读取 ItemReader 中的所有行并将行和文件地址返回给 ItemProcessor？

Question

我在我的批处理 Spring 项目中定义了一个作业流，并定义了 ItemReader、ItemProcessor、ItemWriter 等

我的 ItemReader 如下代码:

@Component
@StepScope
public class MyFileReader extends FlatFileItemReader<FileInfo> {
    private String fileName;

    public MyFileReader () {
    }

    @Value("#{jobParameters[fileName]}")
    public void setFileName(final String fileName) {
        this.fileName = fileName;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Resource resource = new FileSystemResource(fileName);
        setResource(resource);
        setEncoding("UTF-8");
        super.afterPropertiesSet();
    }
}

我的文件输入格式是:

111111,11111,111,111  
222222,22222,222,222  

我想读取文件的所有行并将行和文件地址返回给ItemProcessor，但是FlatFileItemReader 逐行读取。我该怎么做才正确？重写 doRead 方法并手动处理问题是否正确？

Answer 1

如果我理解这个问题，您想从文件中读取所有行，将该数据存储在一个对象中，然后将该对象传递给处理器。一种方法是在作业开始之前使用作业监听器读取文件中的所有行。如下图所示，您可以读取所有行，填充一个表示单行内容的 Java 对象，收集所有这些对象(因此，如果有两行，您将填充 2 个 bean)，然后将它们传递给一次处理一个(如果您愿意，也可以同时处理)。它看起来像这样:

1. 首先您要创建一个监听器。

   public class MyJobListenerImpl implements JobExecutionListener {
       private MyFileReader reader;
   
   @Override
   public void beforeJob(JobExecution jobExecution) {
       reader.init();
   }
   
   @Override
   public void afterJob(JobExecution jobExecution) {
       // noop
   }
   
   // Injected
   public void setReader(MyFileReader reader) {
       this.reader = reader;
   }
   

2. 接下来为您的自定义阅读器添加一个初始化方法。

   public void init() {
       if(Files.exists(inputFileLocation)) {
           List<String> inputData = null;
           try {
               inputData = Files.readAllLines(inputFileLocation, StandardCharsets.UTF_8);
           } catch(IOException e) {
               System.out.println("issue reading input file {}. Error message: {}", inputFileLocation, e);
               throw new IllegalStateException("could not read the input file.");
           }
   
           try {
               for(String fileItem : inputData) {
   
                   YourFileDataBean fileData = new YourFileDataBean();
   
                   yourFileDataBean.setField1(fileItem.split(",")[0].trim());
                   yourFileDataBean.setFiled2(fileItem.split(",")[1].trim());
                   yourFileDataBean.setField3(fileItem.split(",")[2].trim());
                   yourFileDataBean.setField4(fileItem.split(",")[3].trim());
                   myDeque.add(yourFileDataBean);  // really up to you how you want to store your bean but you could add a Deque instance variable and store it there.
   
           }
       } catch(ArrayIndexOutOfBoundsException e) {
           LOGGER.warn("ArrayIndexOutOfBoundsException due to data in input file.");
           throw new IllegalStateException("Failure caused by init() method. Error reading in input file.");
       }
   } else {
       LOGGER.warn("Input file {} does not exist.", inputFileLocation);
       throw new IllegalStateException("Input file does not exist at file location " + inputFileLocation);
   }
   }
   

3. 让您的自定义阅读器中的 read()(或 MyFileReader())方法返回由读入的所有文件行填充的对象。在这个示例中，我正在实现 ItemReader 而不是像您所做的那样扩展它，但你明白了。如果您打算返回代表整个文件的单个 Java 对象，则无需将该对象存储在 Deque 或 List 中。

   @Override
   public MyFileReader read() throws NonTransientResourceException {
   
       return myDeque.poll();
   }
   

   希望这对您有所帮助。

至于将文件地址返回给ItemProcessor。您可以将其作为 YourFileDataBean 中的一个字段并将 inputFileLocation 存储在那里，或者将其保存到执行上下文并以这种方式访问​​它。如果您将此文件路径注入(inject)您的阅读器，您可以在您的处理器中执行相同的操作，假设您的阅读器在确定文件路径方面没有任何作用(又名，它是预先确定的)。