Spring Batch中使用MultiResourceItemReader时如何在读取文件前查询MongoDB

要在每个CSV文件读取前执行MongoDB查询，你可以利用Spring Batch中MultiResourceItemReader提供的资源回调机制，这个机制会在切换到新的文件资源时触发逻辑，完全匹配你针对单个文件执行前置操作的需求，具体实现如下：

1. 封装MongoDB查询逻辑 #

先把前置查询的逻辑封装成独立组件，便于解耦和复用：

@Component
public class PreFileMongoQueryHandler {

    private final MongoTemplate mongoTemplate;

    // 构造注入MongoTemplate
    public PreFileMongoQueryHandler(MongoTemplate mongoTemplate) {
        this.mongoTemplate = mongoTemplate;
    }

    public void executePreQuery(Resource fileResource) {
        // 根据文件资源获取文件名，作为查询条件（可根据你的业务调整）
        String fileName = fileResource.getFilename();
        
        // 执行MongoDB查询逻辑
        Query query = new Query(Criteria.where("fileIdentifier").is(fileName));
        FileRelatedConfig config = mongoTemplate.findOne(query, FileRelatedConfig.class);
        
        // 这里可以根据业务需求处理查询结果，比如缓存、存入执行上下文等
    }
}

2. 配置MultiResourceItemReader的资源回调 #

修改你原有的multiResourceReader Bean配置，添加resourceCallback，将前置查询逻辑绑定到每个文件的处理前：

@Bean
public MultiResourceItemReader<Workout> multiResourceReader(FlatFileItemReader<Workout> reader,
                                                             PreFileMongoQueryHandler queryHandler) {
    return new MultiResourceItemReaderBuilder<Workout>()
            .name("multiWorkoutReader")
            .resources(loadWorkoutFiles())
            .delegate(reader)
            // 每个文件读取前触发的回调
            .resourceCallback(resource -> queryHandler.executePreQuery(resource))
            .build();
}

3. 可选：传递查询结果到后续步骤 #

如果需要将查询结果传递给后续的ItemProcessor或ItemWriter，可以借助ExecutionContext存储数据：

在回调中存入上下文 #

.resourceCallback(resource -> {
    // 获取当前步骤的执行上下文
    StepExecution stepExecution = StepSynchronizationManager.getContext().getStepExecution();
    ExecutionContext executionContext = stepExecution.getExecutionContext();
    
    // 执行查询并获取结果
    FileRelatedConfig config = queryHandler.executePreQuery(resource);
    // 将结果存入上下文
    executionContext.put("currentFileConfig", config);
})

在后续组件中获取上下文数据 #

比如在ItemProcessor中通过@BeforeStep注解获取：

@Component
public class WorkoutProcessor implements ItemProcessor<Workout, Workout> {

    private FileRelatedConfig currentFileConfig;

    @BeforeStep
    public void init(StepExecution stepExecution) {
        this.currentFileConfig = (FileRelatedConfig) stepExecution.getExecutionContext().get("currentFileConfig");
    }

    @Override
    public Workout process(Workout item) throws Exception {
        // 使用currentFileConfig处理当前文件的每条数据
        return item;
    }
}

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