如何为MultiResourceItemReader的每个资源使用ItemReadListener？

问题根源 #

你当前的实现中，MultiResourceItemReader会复用同一个FlatFileItemReader实例和ItemReadListener实例处理所有输入资源，每次调用afterRead时都会用新的customProperty覆盖jobExecutionContext中同一个key的值，最终只会保留最后一个文件的数据。

解决方案 #

以下三种方案可以解决这个问题，根据你的业务需求选择：

方案一：跟踪当前资源，用资源标识作为存储key #

自定义一个包装类跟踪FlatFileItemReader当前处理的资源，在Listener中用资源文件名作为key存储属性，避免覆盖。

1. 自定义资源跟踪Reader：

public class ResourceTrackingFlatFileItemReader<T> extends FlatFileItemReader<T> {
    private Resource currentResource;

    @Override
    public void setResource(Resource resource) {
        this.currentResource = resource;
        super.setResource(resource);
    }

    public Resource getCurrentResource() {
        return currentResource;
    }
}

2. 修改Reader Bean定义：

@Bean
public ResourceTrackingFlatFileItemReader<CustomObject> reader() {
    ResourceTrackingFlatFileItemReader<CustomObject> reader = new ResourceTrackingFlatFileItemReader<>();
    reader.setLineMapper(customObjectLineMapper());       
    return reader;
}

3. 修改Listener逻辑：

public class CustomObjectListener implements ItemReadListener<CustomObject> {

    @Value("#{jobExecution.executionContext}")
    private ExecutionContext executionContext;
    
    @Autowired
    private ResourceTrackingFlatFileItemReader<CustomObject> reader;
    
    @Override
    public void beforeRead() {
    }

    @Override
    public void afterRead(CustomObject item) {
        Resource currentResource = reader.getCurrentResource();
        try {
            // 用文件名作为key，避免属性覆盖
            String resourceKey = "customProperty_" + currentResource.getFilename();
            executionContext.put(resourceKey, item.getCustomProperty());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onReadError(Exception ex) {
    }
}

方案二：使用PartitionStep实现资源隔离处理 #

将每个输入资源作为独立分区，每个分区拥有独立的StepExecutionContext和Listener，从根本上避免属性覆盖。

1. 定义处理单个资源的Step：

@Bean
public Step processSingleResourceStep() {
    return stepBuilderFactory.get("processSingleResourceStep")
            .<CustomObject, CustomObject>chunk(10)
            .reader(reader())
            .writer(new NoOpItemWriter())
            .listener(customObjectListener())
            .build();
}

2. 配置资源分区器：

@Bean
public Partitioner resourcePartitioner() {
    MultiResourcePartitioner partitioner = new MultiResourcePartitioner();
    partitioner.setResources(resources);
    partitioner.setKeyName("inputResource");
    return partitioner;
}

3. 定义主Step为PartitionStep：

@Bean
public Step loadInputFiles() {
    return stepBuilderFactory.get("loadInputFiles")
            .partitioner("processSingleResourceStep", resourcePartitioner())
            .step(processSingleResourceStep())
            .gridSize(resources.length)
            .build();
}

4. 修改Reader为StepScope，从分区上下文获取当前资源：

@Bean
@StepScope
public FlatFileItemReader<CustomObject> reader(@Value("#{stepExecutionContext['inputResource']}") Resource resource) {
    FlatFileItemReader<CustomObject> reader = new FlatFileItemReader<>();
    reader.setResource(resource);
    reader.setLineMapper(customObjectLineMapper());       
    return reader;
}

5. 修改Listener使用StepExecutionContext：

public class CustomObjectListener implements ItemReadListener<CustomObject> {

    @Value("#{stepExecution.executionContext}")
    private ExecutionContext executionContext;
    
    @Override
    public void beforeRead() {
    }

    @Override
    public void afterRead(CustomObject item) {
        executionContext.put("customProperty", item.getCustomProperty());
    }

    @Override
    public void onReadError(Exception ex) {
    }
}

方案三：修改存储结构，收集所有资源属性 #

将jobExecutionContext中的存储结构改为集合（如Map），统一收集所有资源的属性，而不是单个值。

修改Listener逻辑：

public class CustomObjectListener implements ItemReadListener<CustomObject> {

    @Value("#{jobExecution.executionContext}")
    private ExecutionContext executionContext;
    
    @Autowired
    private ResourceTrackingFlatFileItemReader<CustomObject> reader;
    
    @Override
    public void beforeRead() {
        // 初始化属性集合
        if (!executionContext.containsKey("customPropertiesMap")) {
            executionContext.put("customPropertiesMap", new HashMap<String, String>());
        }
    }

    @Override
    public void afterRead(CustomObject item) {
        Resource currentResource = reader.getCurrentResource();
        try {
            String fileName = currentResource.getFilename();
            Map<String, String> propertiesMap = (Map<String, String>) executionContext.get("customPropertiesMap");
            propertiesMap.put(fileName, item.getCustomProperty());
            // 更新上下文
            executionContext.put("customPropertiesMap", propertiesMap);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onReadError(Exception ex) {
    }
}

方案选择建议 #

• 若仅需收集所有资源的属性，方案一或方案三实现简单，无需修改Step结构；
• 若需对每个资源执行独立的验证或输出逻辑，方案二的PartitionStep更适合，它提供了天然的资源隔离，扩展性更强。

内容的提问来源于stack exchange，提问作者zsp