Spring Batch - 如何使用分区并行读写数据？

Question

我有一个 Spring Batch 应用程序 (3.0.7)，它通过 Spring Boot 启动，它并行读取多个 XML 文件、处理它们，然后针对 Oracle DB “吐出” INSERT 或 UPDATE 语句。

为了并行处理文件，我使用了Partitioner。这项工作工作正常，除了JdbcWriter 似乎只绑定到一个线程。由于我使用的是ThreadPoolTaskExecutor，因此我希望 Step 可以为读取器、处理器和写入器并行运行。但是，似乎 JdbcWriter 总是绑定到Thread-1 （我可以在日志中看到，也可以在分析数据库连接时看到，只有一个连接处于活动状态 - 请注意，我的数据源配置为使用具有 20 个连接的池）。

我已将阅读器、处理器和编写器注释为@StepScope。如何有效地使用 taskExecutor 中的所有配置线程并行读取 AND WRITE？

这是我的配置摘录：

@Bean
public Job parallelJob() throws Exception {
    return jobBuilderFactory.get("parallelJob")
            .start(splitFileStep())
            .next(recordPartitionStep())
            .build();
}

@Bean
public Step recordPartitionStep() {
    return stepBuilderFactory.get("factiva-recordPartitionStep")
            .partitioner(recordStep())
            .partitioner("recordStep", recordPartitioner(null)) <!-- this is used to inject some data from the job context
            .taskExecutor(taskExecutor())
            .build();
}

@Bean
public Step recordStep() {
    return stepBuilderFactory.get("recordStep")
            .<Object, StatementHolderMap>chunk(1000)
            .reader(recordReader(null)) <!-- this is used to inject some data from the job context
            .processor(recordProcessor) <!-- this is @Autowired, and the bean is marked as @StepScope
            .writer(jdbcItemWriter())
            .build();
}

@Bean
@StepScope
public ItemStreamReader recordReader(@Value("#{stepExecutionContext['record-file']}") Resource resource) {
    // THIS IS A StaxEventItemReader
}

@Bean
@StepScope
public JdbcItemWriter jdbcItemWriter() {

    JdbcItemWriter jdbcItemWriter = new JdbcItemWriter();
    jdbcItemWriter.setDataSource(dataSource);
    ...
    return jdbcItemWriter;
}

@Value("${etl.factiva.partition.cores}")
private int threadPoolSize;

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    if (threadPoolSize == 0) {
        threadPoolSize = Runtime.getRuntime().availableProcessors();
    }
    taskExecutor.setMaxPoolSize(threadPoolSize);
    taskExecutor.afterPropertiesSet();

    return taskExecutor;
}

Answer 1

我知道为什么 Spring Batch 没有使用所有配置的线程。

首先，Partitioner 的 Spring 配置错误。 原始配置未设置gridSize 值，并且错误地引用了要在分区中运行的步骤。

其次，原配置中使用的ThreadPoolTaskExecutor 似乎不起作用。切换到SimpleAsyncTaskExecutor 就成功了。

我仍然不确定为什么 ThreadPoolTaskExecutor 不起作用。 SimpleAsyncTaskExecutor 的 javadoc 实际上建议使用池来重用线程。

我也不是 100% 确定我完全理解设置 gridSize 值的含义。目前，我将gridSize 设置为等于使用的线程数的值 在分区步骤中。如果有人可以评论这种方法@Michael Minella，那就太好了？ :)

这是正确的配置，供参考。

@Bean
public Job parallelJob() throws Exception {
    return jobBuilderFactory.get("parallelJob")
            .start(splitFileStep())
            .next(recordPartitionStep())
            .build();
}

@Bean
public Step recordPartitionStep() {
    return stepBuilderFactory.get("factiva-recordPartitionStep")
            .partitioner(recordStep().getName(), recordPartitioner(null)) <!-- the value for the recordPartitioner constructor is injected at runtime
            .step(recordStep())
            .gridSize(determineWorkerThreads()) <!-- GRID SIZE VALUE MUST BE EQUAL TO THE NUMBER OF THREAD CONFIGURED FOR THE THREAD POOL
            .taskExecutor(taskExecutor())
            .build();


}

@Bean
public Step recordStep() {
    return stepBuilderFactory.get("recordStep")
            .<Object, StatementHolderMap>chunk(1000)
            .reader(recordReader(null)) <!-- this is used to inject some data from the job context
            .processor(recordProcessor) <!-- this is @Autowired, and the bean is marked as @StepScope
            .writer(jdbcItemWriter())
            .build();
}

@Bean
@StepScope
public ItemStreamReader recordReader(@Value("#{stepExecutionContext['record-file']}") Resource resource) {
    // THIS IS A StaxEventItemReader
}

@Bean
@StepScope
public JdbcItemWriter jdbcItemWriter() {

    JdbcItemWriter jdbcItemWriter = new JdbcItemWriter();
    jdbcItemWriter.setDataSource(dataSource);
    ...
    return jdbcItemWriter;
}

@Value("${etl.factiva.partition.cores}")
private int threadPoolSize;

@Bean
public TaskExecutor taskExecutor() {
    SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor("fac-thrd-");

    taskExecutor.setConcurrencyLimit(determineWorkerThreads());
    return taskExecutor;
}

// threadPoolSize is a configuration parameter for the job
private int determineWorkerThreads() {
    if (threadPoolSize == 0) {
        threadPoolSize = Runtime.getRuntime().availableProcessors();
    }
    return threadPoolSize;

}