java - 线程执行器在 Spring 批处理中不起作用

我的治疗持续了9分钟，我想通过使用线程执行器来减少执行时间。

我的阅读器读取数据库中的1200行，然后进行UNPIVOT，这使我可以将56036行写入csv文件。

我尝试了几种方法，但是时间没有差异，给人的印象是没有考虑配置。

@Bean
@StepScope
public ItemReader<DmNebefPdhExportRetenuCSV> datamartEffRetenuItemReader(
        @Value("#{jobParameters['dateExport']}") Date dateExport) throws Exception {
    PagingQueryProvider query = createEffRetenuQuery();
    Map<String, Object> parameters = new HashMap<>();
    parameters.put("dateExport", dateExport);

    JdbcPagingItemReader<DmNebefPdhExportRetenuCSV> reader = new JdbcPagingItemReader<>();
    reader.setDataSource(sdmDataSource);
    reader.setParameterValues(parameters);
    reader.setQueryProvider(query);
    reader.setFetchSize(1000);
    reader.setPageSize(1000);
    reader.setRowMapper(new BeanPropertyRowMapper<>(DmNebefPdhExportRetenuCSV.class));

    return reader;
}

@Bean
@StepScope
FlatFileItemWriter<DmNebefPdhExportRetenuCSV> exportEffRetenuItemWriter(
        @Value("#{jobParameters['dateExport']}") Date dateExport) {
    // Construction du Header
    StringHeaderWriter headerWriter = new StringHeaderWriter(EXPORT_EFF_RETENU_CSV_HEADER);

    String newExportFileVersion = getExportRetenuVersion(dateExport);

    // Nom du fichier d'export
    String csvFileName = createEffRetenuExportFileName(dateExport, newExportFileVersion);

    // Chemin complet d'enregistrement du fichier
    String exportFilePath = String.join("/", exportArchiveCreRetenuPath, csvFileName);

    // Définition du délimiteur et des champs à mapper
    LineAggregator<DmNebefPdhExportRetenuCSV> lineAggregator = effRetenuLineAggregator();

    // Put dans le context pour récupérer dans le listener
    exportEffRetenuJobListener.getJobExecution().getExecutionContext().put("exportFilePath", exportFilePath);
    exportEffRetenuJobListener.getJobExecution().getExecutionContext().put("csvFileName", csvFileName);
    exportEffRetenuJobListener.getJobExecution().getExecutionContext().put("newExportFileVersion",
            newExportFileVersion);

    FlatFileItemWriter<DmNebefPdhExportRetenuCSV> csvFileWriter = new FlatFileItemWriter<>();
    csvFileWriter.setShouldDeleteIfEmpty(true);
    csvFileWriter.setHeaderCallback(headerWriter);
    csvFileWriter.setResource(new FileSystemResource(exportFilePath));
    csvFileWriter.setLineAggregator(lineAggregator);

    return csvFileWriter;

}

@Bean
public Step exportCSVStep() throws Exception {
    return stepBuilderFactory.get("exportCSVStep")
            .<DmNebefPdhExportRetenuCSV, DmNebefPdhExportRetenuCSV>chunk(100)
            .reader(datamartEffRetenuItemReader(WILL_BE_INJECTED))
            .listener(readListener)
            .writer(exportEffRetenuItemWriter(WILL_BE_INJECTED))
            .listener(writeListener)
            .taskExecutor(taskExecutor())
            .throttleLimit(50)
            .build();
}

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor ();
    taskExecutor.setMaxPoolSize(50);
    taskExecutor.afterPropertiesSet();
    return taskExecutor;
}

@Bean
public Job exportEffRetenuJob() throws Exception {
    return jobBuilderFactory
            .get("exportEffRetenuJob")
            .incrementer(new RunIdIncrementer())
            .listener(exportEffRetenuJobListener)
            .flow(exportCSVStep()).end().build();
}

我也尝试过使用新的SimpleAsyncTaskExecutor（）。

将setFetchSize（）和setPageSize（）添加到我的阅读器中，处理时间从9分钟减少到10秒。
但是ThreadExecutor似乎不起作用。

日志：

没有TaskExecutor时：12,751秒

[2018-01-03 11:41:32,087] INFO  ExportEffRetenuJobListener - Start job exportEffRetenuJob - export month 04-2017
[2018-01-03 11:41:44,838] INFO  ExportEffRetenuJobListener - End job : exportEffRetenuJob - export month : 04-2017 - statut : COMPLETED

使用TaskExecutor：11,328秒

[2018-01-03 11:42:55,439] INFO  ExportEffRetenuJobListener - Start job exportEffRetenuJob - export month 04-2017
[2018-01-03 11:43:06,767] INFO  ExportEffRetenuJobListener - End job : exportEffRetenuJob - export month : 04-2017 - statut : COMPLETED

测试24万行以读取/写入

使用TaskExecutor：123秒

没有TaskExecutor时：127秒

我认为我没有很好地配置任务执行器。

最佳答案

您必须为执行者设置一个corePoolSize。默认值为1，因此您仍然只能得到一个线程。尝试设置taskExecutor.setCorePoolSize(50)，然后重新运行作业。