Spring 批处理 : Aggregating records and write count

Question

我们在平面文件中有一些数据。例如

EmpCode,Salary,EmpName,...  
100,1000,...,...
200,2000,...,...
200,2000,...,...
100,1000,...,...
300,3000,...,...
400,4000,...,...

我们想根据 EmpCode 汇总工资并写入数据库

Emp_Code    Emp_Salary   Updated_Time   Updated_User 
100         2000         ...            ...
200         4000         ...            ...
300         3000         ...            ...
400         4000         ...            ...

我已经按照 Spring Batch 编写了如下类

ItemReader - to read the employee data into a Employee object                

示例 EmployeeItemProcessor:

public class EmployeeProcessor implements ItemProcessor<Employee, Employee> {

    @Override
    public Employee process(Employee employee) throws Exception {
        employee.setUpdatedTime(new Date());
        employee.setUpdatedUser("someuser");
        return employee;
    }

EmployeeItemWriter:

@Repository
public class EmployeeItemWriter implements ItemWriter<Employee> { 
 @Autowired
 private SessionFactory sf;

 @Override  
 public void write(List<? extends Employee> employeeList) throws Exception {  
  List<Employee> aggEmployeeList = aggregateEmpData(employeeList);
  //write to db using session factory
 }  

 private List<Employee> aggregateEmpData(List<? extends Employee> employeeList){
     Map<String, Employee> map = new HashMap<String, Employee>(); 
    for(Employee e: employeeList){
        String empCode =  e.getEmpCode();
        if(map.containsKey(empCode)){
            //get employee salary and add up
         }else{
          map.put(empCode,Employee);
         }
     }    
     return new ArrayList<Employee>(map.values());         
 }
}

XML 配置

...
<batch:job id="employeeJob">
    <batch:step id="step1">
    <batch:tasklet>
        <batch:chunk reader="employeeItemReader" 
            writer="employeeItemWriter" processor="employeeItemProcessor"
            commit-interval="100">
        </batch:chunk>
    </batch:tasklet>
    </batch:step>
  </batch:job>
...

它正在发挥作用并为我服务。不过，我有几个问题。

1) 当我查看日志时，它显示如下(commit-interval=100):

status=COMPLETED，exitStatus=COMPLETED，readCount=2652，filterCount=0，writeCount=2652 readSkipCount=0，writeSkipCount=0，processSkipCount=0，commitCount=27，rollbackCount=0

但是聚合之后，只有2515条记录写入了数据库。 write count是2652，是不是因为到达ItemWriter的item数量还是2652？如何纠正？

2) 我们遍历列表两次。一次在 ItemProcessor 中，然后在 ItemWriter 中进行聚合。如果记录数较多，则可能是性能问题。有没有更好的方法来实现这一点？

Answer 1

如果输入文件的每一行都是一个员工对象，那么您的ReadCount 就是输入文件中的行数。 WriteCount 将是传递给项目作者的所有列表的大小总和。因此，也许您的 aggregateEmpData 函数删除或聚合了一些记录为一条记录，因此，您的数据库计数与 WriteCount 不同。 如果您想确保 WriteCount 正好是数据库中的记录数，您应该在处理器中进行聚合。