在这种情况下使用 Tasklet 或 Chunk

Posted 2023-02-27

【中文标题】在这种情况下使用 Tasklet 或 Chunk

【英文标题】：To user Tasklet or Chunk in this scenario

【发布时间】：2021-02-04 11:22:29

【问题描述】：

我有一个工作/任务给read sub-folders/directory 一个given folder/path。路径是动态的，我们从Controller 获取。目前，我使用Tasklet，有3个tasklet，一个读取子目录，一个处理它以准备对象保存到数据库，最后一个将处理后的数据对象写入数据库。 文件夹可以有任意数量的子文件夹。目前，我使用了这个代码：

 Path start = Paths.get("x:\\data\\");
    Stream<Path> stream = Files.walk(start, 1);
List<String> collect = stream
                .map(String::valueOf)
                .sorted()
                .collect(Collectors.toList());

一次读取所有子文件夹。 为此，我遵循了Tasklet 实现的https://www.baeldung.com/spring-batch-tasklet-chunk 示例。这是正确的方法吗？我还需要使用多线程异步运行作业。 由于可能有大量子文件夹，so there can be huge number of rowsorlist of data to process and write to the database.

请提出适当的方法。 我正在学习Spring Batch，在file read/process/write 上也做了一些例子，并为此使用了Chunk 方法。 但我的工作是读取文件夹/路径的子目录，所以我无法决定采用哪种方法。

【参考方案1】：

我有一个类似的场景：我需要从一个文件夹中读取所有文件，在db中处理和写入，(Doc)

@Configuration
@EnableBatchProcessing
public class BatchConfig 

@Bean
public Job job(JobBuilderFactory jobBuilderFactory,
               Step masterStep) 
    return jobBuilderFactory.get("MainJob")
            .incrementer(new RunIdIncrementer())
            .flow(masterStep)
            .end()
            .build();


@Bean
public Step mainStep(StepBuilderFactory stepBuilderFactory,
                     JdbcBatchItemWriter<Transaction> writer,
                     ItemReader<String> reader,
                     TransactionItemProcessor processor) 
    return stepBuilderFactory.get("Main")
            .<String, Transaction>chunk(2)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            **.taskExecutor(jobTaskExecutor())**
            .listener(new ItemReaderListener())
            .build();


@Bean
public TaskExecutor jobTaskExecutor() 
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(2);
    taskExecutor.setMaxPoolSize(10);
    taskExecutor.afterPropertiesSet();
    return taskExecutor;


@Bean
@StepScope
public ItemReader<String> reader(@Value("#stepExecution") StepExecution stepExecution) throws IOException 
    Path start = Paths.get("D:\\test");
    List<String> inputFile = Files.walk(start, 1)
            .map(String::valueOf)
            .sorted()
            .collect(Collectors.toList());
    return new IteratorItemReader<>(inputFile);


@Bean
@StepScope
public TransactionItemProcessor processor(@Value("#stepExecution") StepExecution stepExecution) 

    return new TransactionItemProcessor();


@Bean
@StepScope
public JdbcBatchItemWriter<Transaction> writer(DataSource dataSource) 

    return new JdbcBatchItemWriterBuilder<Transaction>()
            .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
            .sql("INSERT INTO transaction (id, date, type) VALUES (:id, :date, :type)")
            .dataSource(dataSource)
            .build();

【讨论】：

,谢谢您的回复。在我的上下文中，我只需要获取子文件夹的路径，不需要读取文件。例如，主文件夹 = //a 并且有子文件夹 /a/b、/a/c、/a/d、.等，并创建对象列表包含要保存到 DB 的父路径和子文件夹路径.所以我的问题是，如果我按照您的方法，它会一次读取所有子文件夹还是逐块读取？您的示例是基于块的方法。如果我们想在 Tasklet 方法中执行此操作，该怎么做？您能否修改示例以仅读取路径，如我的代码 sn-p 中还包括阅读器？

经过一些测试，我认为最好的方法是使用块，我编辑了代码。添加 TaskExecutor 允许您添加多线程，如果您想要更好的性能，您可以检查远程分块。我编辑了代码

TaskExecutor 集合在哪里？

在mainStep中