Spring Batch AsyncItemProcessor 不并行处理答案

【问题标题】：Spring Batch AsyncItemProcessor not processing in parallelSpring Batch AsyncItemProcessor 不并行处理
【发布时间】：2023-02-08 01:58:32
【问题描述】：

我有一个 AsyncItemProcessor，我想与以下配置并行运行。但是，处理不是并行发生的。


@Configuration
@EnableBatchProcessing
@EnableAsync
public class JobConfig {
   @Autowired
    private JobBuilderFactory jobBuilder;

    @Autowired
    private StepBuilderFactory stepBuilder;

    @Autowired
    @Qualifier("writer")
    private ItemWriter writer;

    @Bean
    @JobScope
    public ItemProcessor itemProcessor() {
        ItemProcessor itemProcessor = new ItemProcessor();
        return itemProcessor;
    }

    @Bean
    @JobScope
    public AsyncItemProcessor asyncItemProcessor() throws IOException {
        AsyncItemProcessor asyncItemProcessor = new AsyncItemProcessor();
        asyncItemProcessor.setDelegate(itemProcessor());
        asyncItemProcessor.setTaskExecutor(getAsyncExecutor());
        asyncItemProcessor.afterPropertiesSet();
        return asyncItemProcessor;
    }

    @Bean(name = "asyncExecutor")
    public TaskExecutor getAsyncExecutor() {
        SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor() {
            @Override
            protected void doExecute(Runnable task) {
                final JobExecution jobExecution = JobSynchronizationManager.getContext().getJobExecution();
                super.doExecute(() -> {
                    JobSynchronizationManager.register(jobExecution);
                    try {
                        task.run();
                    } finally {
                        JobSynchronizationManager.close();
                    }
                });
            }
        };
        simpleAsyncTaskExecutor.setThreadNamePrefix("processing 1-");
        simpleAsyncTaskExecutor.setConcurrencyLimit(100);
        return simpleAsyncTaskExecutor;
    }

  

    @Bean
    @JobScope
    public AsyncItemWriter asyncItemWriter() throws IOException {
        AsyncItemWriter asyncItemWriter = new AsyncItemWriter<>();
        asyncItemWriter.setDelegate(writer);
        asyncItemWriter.afterPropertiesSet();
        return asyncItemWriter;
    }

    @Bean
    public Step simpleFileStep() throws IOException {
        return stepBuilder.get("simpleFileStep").chunk(100).reader(fileReader).processor(asyncItemProcessor())
                .writer(asyncItemWriter()).taskExecutor(getAsyncExecutor()).build();
    }
}

在这种情况下，我想并行处理包含 1000 条记录的文件，但一次只处理和写入 100 个项目。如果配置有问题，请告诉我。

【问题讨论】：

我添加了你的问题的答案。我仍然很想知道覆盖SimpleAsyncTaskExecutor#doExecute 并用JobSynchronizationManager.register 和JobSynchronizationManager.close 包装任务执行的目的。

标签： spring spring-batch spring-batch-tasklet

【解决方案1】：

并发不同于并行。 AsyncItemProcessor 旨在与AsyncItemWriter 携手合作，同时处理项目。在您的情况下，将同时处理 100 个项目的单个块，但不会并行处理块。它仍然是块的串行执行，但是每个块都由任务执行器的不同线程并发处理。

在 Spring Batch 中无法并行处理块。 Spring Batch 提供的是分区，分区可以并行处理（使用本地线程或远程 JVM）。

因此，您可以做的是对输入进行分区（例如，每个分区中有 1000 个项目）并配置分区步骤以并行处理分区。请注意，每个分区也可以同时进一步处理块。您可以在此处的参考文档中找到更多详细信息和代码示例：Scaling and Parallel Processing。

【讨论】：