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

Question

我有一个包含Reader->Processor->Writer的spring批处理。

B/W传递的数据类型为emp:

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

    // getters and setters
}

在读取器中从CSV文件读取一个简单的批处理数据时，处理器中的一些处理&输出CSV文件由写入器编写。

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;
        }

}

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

谢谢

Answer 1

为此，您可以使用ItemWriteListener和JobExecutionListenerSupport。

>

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

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

@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();
}