Spring Batch - 带有列表的Item Reader和ItemProcessor

Question

我正在使用一个弹簧批处理应用程序，它读取一个Flat文件并返回一个对象。我想做的就是让FlatFileItemReader返回一个列表并将其传递给处理器，以便它将每个列表视为一个项目。请参阅下面的代码段

@Bean public FlatFileItemReader <List<T>> reader() throws Exception {
    //reader.read()
    }

 @Bean
    public ItemProcessor <List<T>, V> getTargetValueProcessor() {
        return new ItemProcessor <List<T>, V>() {
            @Override
            public V process(List<T> t) throws Exception {
                //processing logic
            }
        };   }

但我的物品处理器将列表中的每个项目视为处理器的单个输入，处理器称为列表大小的次数。如果阅读器返回的列表大小为3，则处理器被调用三次。关于如何在ItemProcessor中处理列表输入的任何想法？

TIA

Answer 1

您可以在这里找到一个项目阅读器的示例，它将List对象作为单个项目返回：https://github.com/spring-projects/spring-batch/tree/master/spring-batch-samples#multiline

这是实施：https://github.com/spring-projects/spring-batch/blob/master/spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/multiline/AggregateItemReader.java#L55

Answer 2

您可以编写如下所示的通用类，它可以用于一次读取多行并将这些作为List<T>传递给处理器，希望它有帮助:)

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.NonTransientResourceException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.ResourceAwareItemReaderItemStream;
import org.springframework.core.io.Resource;

public class FlatFileItemGroupReader<T> implements ResourceAwareItemReaderItemStream<List<T>> {

private int groupSize = 100;

private static final Logger LOG = LogManager.getLogger(FlatFileItemGroupReader.class);

private FlatFileItemReader<T> fileReader;

@Override
public synchronized List<T> read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
    List<T> records = new ArrayList<T>();

    while(records.size() < groupSize) {
        T record = fileReader.read();
        if (Objects.isNull(record)) {
            break;
        }
        records.add(record);
    }

    if (records.isEmpty()) {
        return null;
    }

    return records;
}

@Override
public synchronized void open(ExecutionContext executionContext) throws ItemStreamException {
    fileReader.open(executionContext);
}

@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
}

@Override
public synchronized void close() throws ItemStreamException {
    fileReader.close();
}

public FlatFileItemReader<T> getFileReader() {
    return fileReader;
}

public void setFileReader(FlatFileItemReader<T> fileReader) {
    this.fileReader = fileReader;
}

public int getGroupSize() {
    return groupSize;
}

public void setGroupSize(int groupSize) {
    this.groupSize = groupSize;
}

@Override
public void setResource(Resource resource) {
    this.fileReader.setResource(resource);

}
}