Spring Batch - 在处理器和写入器之间传递数据

Question

我有一个包含 reader->processor->writer 的 spring batch。

b/w 传递的数据类型为Emp：

class Emp {
    iny id;
    String name;
    EmpTypeEnum empType;    // HR, Dev, Tester, etc.

    // getters and setters
}

由于从Reader 中的CSV 文件读取简单的批处理数据，Processor & 内部的一些处理和输出 CSV 文件由Writer 写入。

但除了这个输出 CSV 文件，我想生成一个辅助输出文件，它只包含每个 EmpType 的计数，即 HR、Dev 和 Tester 的总数。

我正在考虑仅在 processor 内执行计数，例如：

public class EmpItemProcessor implements ItemProcessor<Emp, Emp> {

    int countHr;
    int countDev;
    int countTester;


        @Override
        public Person process(final Emp emp) throws Exception {

        if (item.getEmpType.equals(EmpTypeEnum.HR) {
            countHr++;
        } else if // .....

        // other processor on emp

            return emp;
        }

}

但正如你所见，我只能从Processor 返回Emp，那么我怎样才能从处理器传递countHr、countDev 等并使用它来创建辅助文件？ p>

请提出建议。如果您认为任何其他方法会更好，请提出建议。

谢谢

Answer 1

您可以为此使用ItemWriteListener 和JobExecutionListenerSupport。

1. 定义一个 ItemWriteListener ，每次调用你的 writer 后都会调用它。

2. 在此监听器中，每次更新执行上下文中的计数器

3. 编写一个 JobExecutionListener，它将在整个作业完成后调用，您可以在其中从执行上下文中读取值并进行进一步处理。

   @Component
   @JobScope
   public class EmployeeWriteListener implements ItemWriteListener<Emp> {
   
     @Value("#{jobExecution.executionContext}")
     private ExecutionContext executionContext;
   
   
     @Override
     public void afterWrite(final List<? extends Emp> paramList) {
   
        final int counter =
             this.executionContext.getInt("TOTAL_EXPORTED_ITEMS", 0);
         this.executionContext.putInt("TOTAL_EXPORTED_ITEMS", counter + 1);
       }
   
     }
   }
   
   
   
   @Component
   @JobScope
   public class EmployeeNotificationListener extends JobExecutionListenerSupport  {
   
   @Override
   public void afterJob(final JobExecution jobExecution) {
   
     jobExecution.getExecutionContext()
         .getInt("TOTAL_EXPORTED_ITEMS")
     ...................
      }
    }
   

你应该在声明你的步骤和工作时注册这些监听器。

this.jobBuilders.get("someJob").incrementer(new RunIdIncrementer()).listener(new EmployeeNotificationListener())
        .flow(this.getSomeStep()).end().build();
//instead of new(..) you should Autowire listener

public Step getSomeStep() {

  return stepBuilders.get("someStep").<X, Y>chunk(10)
      .reader(this.yourReader).processor(this.yourProcessor)
      .writer(this.yourProcessor).listener(this.EmployeeWriteListener)
      .build();
}

Answer 2

基本上你需要多个ItemWriter 来处理两个不同的写作任务。您可以轻松使用CompositeItemWriter，它可以在其中保存不同 ItemWriter 的列表。在每个项目上，它都会调用它的所有 ItemWriter。

在你的情况下，

1. 制作两个FlatFileItemWriter - 一个用于您的正常 CSV 输出，另一个用于您的统计信息。

2. 然后创建 CompositeItemWriter<Emp> 对象并使用它的这种方法将这两个 FlatFileItemWriter<Emp> 添加到其中 - public void setDelegates(List<ItemWriter<Emp>> delegates)

3. 在步骤中使用这个CompositeItemWriter 就像你ItemWriter 一样

因此，当您的 CompositeItemWriter 被调用时，它将委托给 ItemWriter，以便您添加到列表中。

工作完成:)