Spring批处理并行处理根据step1结果创建步骤答案

【问题标题】：Spring batch parallel processing create steps based on the step1 resultSpring批处理并行处理根据step1结果创建步骤
【发布时间】：2019-07-30 02:38:22
【问题描述】：

我正在尝试使用 Spring Boot 和 Spring Batch 实现并行处理。此批次将通过一些必需参数从 UI 触发

我需要根据请求参数创建步骤，我尝试如下，

其余控制器看起来，

JobParameters jobParameters = new JobParametersBuilder().addLong("JobID",System.currentTimeMillis())
                    .addString("fileName", filename) 
                    .addString("buisinessDate", model.getGeneralServiceModel().getBusinessDate()) 
                    .addString("source", model.getGeneralServiceModel().getSource()) 
                    .toJobParameters();
            jobLauncher.run(job, jobParameters);

还有批处理配置：

Flow masterFlow = (Flow)new FlowBuilder("masterFlow").start(stepOne()).build();

    List<Step> steps = new ArrayList<Step>();
    for (ConcurrentLinkedQueue date : taskOne.readFile()) {
        steps.add(createStep(date));
    }

    return jobs.get("myJob")
            .start(masterFlow)
             .next(createParallelFlow(steps))
             .end()
             .build();

masterFlow 将作业参数读入其变量中，readFile() 给出列表（基于此，必须创建步骤），为此需要 jobParameters。

问题是：

在启动我的应用程序本身时，readFile() 正在执行。但是我需要在作业通过 RestController 触发时执行它，因为它具有所需的参数。

如何在启动应用程序时停止此执行？

【问题讨论】：

不清楚你在问什么。在标题中你说create steps based on the step1 result，在问题中你说I need to create steps based on the request parameters。这些是不同的东西。请澄清您的问题并举一个带有真实参数的示例，以了解您要实现的目标。
我需要根据 step1 结果创建步骤。但为了处理第 1 步，我需要请求参数。

标签： multithreading spring-boot parallel-processing spring-batch tasklet

【解决方案1】：

我需要根据 step1 结果创建步骤

创建步骤是您在配置时执行的操作。 Step1 结果只能在运行时知道。因此，为了做到这一点，您需要在运行时访问应用程序上下文并根据 Step1 的结果动态注册 step bean。我不确定这是否真的是您的目标。

但是，如果您想根据 step1 的结果执行（不是创建，而是执行）这些步骤，那么您可以使用JobExecutionDecider。请参阅Programmatic Flow Decisions 了解更多详细信息和代码示例。可以在这里找到类似的问题：How to use decider in Spring batch?

但为了处理第 1 步，我需要请求参数。

我看到您已经设法获取请求参数并将它们设置为作业参数。您可以做的是通过步骤执行访问您的步骤中的那些作业参数。如果您的步骤是一个简单的 Tasklet，这里是一个示例：

class MyTasklet implements Tasklet {
    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        Map<String, Object> jobParameters = chunkContext.getStepContext().getJobParameters();
        // use job parameters
        return RepeatStatus.FINISHED;
    }
}

如果您的步骤是面向块的 tasklet，您可以使用 StepListener#beforeStep 来访问步骤执行。

希望这会有所帮助。

【讨论】：