Spring 批处理 - 如何根据上一步创建的参数生成并行步骤

Question

简介

我正在尝试使用在 tasklet 中创建的作业参数在 tasklet 执行后创建步骤。

tasklet 尝试查找一些文件 (findFiles())，如果它找到一些文件，它会将文件名保存到字符串列表中。

在 tasklet 中，我按如下方式传递数据： chunkContext.getStepContext().getStepExecution().getExecutionContext().put("files", fileNames);

下一步是并行流程，其中每个文件都会执行一个简单的 reader-processor-writer 步骤（如果您对我如何到达那里感兴趣，请参阅我之前的问题：)

在构建作业 readFilesJob() 时，最初使用 "fake" 文件列表创建一个流，因为只有在执行了 tasklet 之后，真正的文件列表才是已知的。

问题

如何配置我的作业，以便首先执行 tasklet，然后使用从 tasklet 生成的文件列表执行并行流？

我认为这归结为在运行时的正确时刻获取加载了正确数据的文件名列表...但是如何？

复制

这是我的简化配置：

@Configuration
@EnableBatchProcessing
public class BatchConfiguration {

    private static final String FLOW_NAME = "flow1";
    private static final String PLACE_HOLDER = "empty";

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    public List<String> files = Arrays.asList(PLACE_HOLDER);

    @Bean
    public Job readFilesJob() throws Exception {   
        List<Step> steps = files.stream().map(file -> createStep(file)).collect(Collectors.toList());

        FlowBuilder<Flow> flowBuilder = new FlowBuilder<>(FLOW_NAME);

        Flow flow = flowBuilder
                .start(findFiles())             
                .next(createParallelFlow(steps))
                .build();       

        return jobBuilderFactory.get("readFilesJob")                
                .start(flow)                
                .end()
                .build();
    }

    private static Flow createParallelFlow(List<Step> steps){
        SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
        taskExecutor.setConcurrencyLimit(steps.size());

        List<Flow> flows = steps.stream()
                .map(step ->
                        new FlowBuilder<Flow>("flow_" + step.getName()) 
                        .start(step) 
                        .build()) 
                .collect(Collectors.toList());

        return new FlowBuilder<SimpleFlow>("parallelStepsFlow").split(taskExecutor) 
             .add(flows.toArray(new Flow[flows.size()]))
             .build();      
    }

    private Step createStep(String fileName){
        return stepBuilderFactory.get("readFile" + fileName)
                .chunk(100)
                .reader(reader(fileName))               
                .writer(writer(filename))                               
                .build();
    }

    private FileFinder findFiles(){
        return new FileFinder();
    }
}

研究

How to safely pass params from Tasklet to step when running parallel jobs 的问题和答案建议在 reader/writer 中使用这样的结构：

@Value("#{jobExecutionContext[filePath]}") String filePath

但是，由于在 createParallelFlow() 方法中创建步骤的方式，我真的希望可以将文件名作为字符串传递给 reader/writer。因此，即使该问题的答案可能是我这里的问题的解决方案，它也不是理想的解决方案。但是，如果我错了，请不要纠正我。

关闭

我正在使用文件名示例来更好地阐明问题。我的问题实际上不是从目录中读取多个文件。我的问题实际上归结为在运行时生成数据并将其传递给下一个动态生成的步骤的想法。

编辑：

添加了 fileFinder 的简化 tasklet。

@Component
public class FileFinder implements Tasklet, InitializingBean {

    List<String> fileNames;

    public List<String> getFileNames() {
        return fileNames;
    }

    @PostConstruct
    public void afterPropertiesSet() {
        // read the filenames and store dem in the list
        fileNames.add("sample-data1.csv");
        fileNames.add("sample-data2.csv");
    }

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        // Execution of methods that will find the file names and put them in the list...
        chunkContext.getStepContext().getStepExecution().getExecutionContext().put("files", fileNames);                     
        return RepeatStatus.FINISHED;
    }    
}

Answer 1

我不确定，如果我没有正确理解你的问题，但据我所知，你需要在之前拥有文件名列表你建立你的工作动态地。

你可以这样做：

@Component
public class MyJobSetup {
    List<String> fileNames;

    public List<String> getFileNames() {
        return fileNames;
    }

    @PostConstruct
    public void afterPropertiesSet() {
        // read the filenames and store dem in the list
        fileNames = ....;
    }
}

之后，您可以将此 Bean 注入到您的 JobConfiguration Bean 中

@Configuration
@EnableBatchProcessing
@Import(MyJobSetup.class)
public class BatchConfiguration {

    private static final String FLOW_NAME = "flow1";
    private static final String PLACE_HOLDER = "empty";

    @Autowired
    private  MyJobSetup jobSetup; // <--- Inject
          // PostConstruct of MyJobSetup was executed, when it is injected

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    public List<String> files = Arrays.asList(PLACE_HOLDER);

    @Bean
    public Job readFilesJob() throws Exception {   
        List<Step> steps = jobSetUp.getFileNames() // get the list of files
             .stream() // as stream
             .map(file -> createStep(file)) // map...
             .collect(Collectors.toList()); // and create the list of steps

Spring 批处理 - 如何根据上一步创建的参数生成并行步骤

Spring Batch - How to generate parallel steps based on params created in a previous step

java

spring-batch

jobs

spring-java-config

late-binding

简介

问题

复制

研究

关闭

编辑：