如何在spring批处理中使用MultiResourceItemReader在读取单个csv后指定taskexecutor生成线程

Question

尝试使用MultiResourceItemReader读取spring批处理中的多个文件，并为每个文件中的记录提供taskExecutor以在多线程中读取。假设一个文件夹中有3个csv文件，MultiResourceItemReader应该逐个执行它，但由于我有taskExecutor，不同的线程占用csv文件，就像两个csv文件被来自同一文件夹的线程占用并开始执行一样。

期望：-MultiResourceItemReader应该读取第一个文件，然后taskExecutor应该生成不同的线程并执行。然后应该拾取另一个文件,并由taskExecutor进行执行。

代码段/batch_configuration:-@bean public Step Step1(){return stepBuilderFactory.get（“Step1”）。 chunk(5)。reader（multiResourceItemReader（））。writer（writer（））。taskExecutor（taskExecutor（））。throttleLimit（throttleLimit）。build（）；}

@Bean
public MultiResourceItemReader<POJO> multiResourceItemReader() 
{
    MultiResourceItemReader<POJO> resourceItemReader = new MultiResourceItemReader<POJO>();
    ClassLoader cl = this.getClass().getClassLoader();
    ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver(cl);

    Resource[] resources;
    try {
        resources = resolver.getResources("file:/temp/*.csv");
         resourceItemReader.setResources(resources);
            resourceItemReader.setDelegate(itemReader());
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    return resourceItemReader;
}

Answer 1

也许您应该检查Partitioner(https://docs.spring.io/spring-batch/docs/current/reference/html/scalability.html#partitioning)并实现如下所示：

@Bean
public Step mainStep(StepBuilderFactory stepBuilderFactory,
                     FlatFileItemReader itemReader,
                     ListDelegateWriter listDelegateWriter,
                     BatchProperties batchProperties) {
    return stepBuilderFactory.get(Steps.MAIN)
            .<POJO, POJO>chunk(pageSize)
            .reader(unmatchedItemReader)
            .writer(listDelegateWriter)
            .build();
}

@Bean
public TaskExecutor jobTaskExecutor(@Value("${batch.config.core-pool-size}") Integer corePoolSize,
                                    @Value("${batch.config.max-pool-size}") Integer maxPoolSize) {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(corePoolSize);
    taskExecutor.setMaxPoolSize(maxPoolSize);
    taskExecutor.afterPropertiesSet();
    return taskExecutor;
}

@Bean
@StepScope
public Partitioner partitioner(@Value("#{jobExecutionContext[ResourcesToRead]}") String[] resourcePaths,
                               @Value("${batch.config.grid-size}") Integer gridSize) {
    Resource[] resourceList = Arrays.stream(resourcePaths)
            .map(FileSystemResource::new)
            .toArray(Resource[]::new);
    MultiResourcePartitioner partitioner = new MultiResourcePartitioner();
    partitioner.setResources(resourceList);
    partitioner.partition(gridSize);
    return partitioner;
}

@Bean
public Step masterStep(StepBuilderFactory stepBuilderFactory, Partitioner partitioner, Step mainStep, TaskExecutor jobTaskExecutor) {
    return stepBuilderFactory.get(BatchConstants.MASTER)
            .partitioner(mainStep)
            .partitioner(Steps.MAIN, partitioner)
            .taskExecutor(jobTaskExecutor)
            .build();
}