如何定义spring batch中另一个流中的并行子流？

Question

我想在spring batch中实现如下的流结构。

           Job
          /   \
       Flow1  Flow2  
         /      \
      Step1    Step2
        /       /  \
       /    Step3  Flow3
      /                \  
     /                 Step4
     \                  /
      \                /
       \              /    
            Step5    

作业配置伪代码如下：

@Configuration
public class JobConfiguration {

......

    @Bean
    public Job Job() {

    Flow flow1 = new FlowBuilder<SimpleFlow>("flow1")
                            .start(step1())
                            .build();

    Flow flow2 = new FlowBuilder<SimpleFlow>("flow2")
           .start(step2())
           .next(step3())
           .split(new SimpleAsyncTaskExecutor()).add(flow3)
           .build();

    Flow flow3 = new FlowBuilder<SimpleFlow>("flow3")
                            .start(step4())
                            .build();

    return jobBuilderFactory.get("job")
            .incrementer(new RunIdIncrementer())
            .start(flow1)
            .split(new SimpleAsyncTaskExecutor()).add(flow2)
            .next(Step5())
            .end()
            .build();
    }
......
}

当我运行批处理时，日志显示执行了步骤1、步骤2、步骤3和步骤5，但没有运行步骤4。

我想知道如何在另一个流中定义子流，上面的代码是实现它的正确方法吗？

提前谢谢！

Answer 1

单独运行每个流表明Flow1和Flow3是正确的，但Flow2不是正确的。只运行Flow2：

    return jobBuilderFactory.get("job")
            .incrementer(new RunIdIncrementer())
            .start(flow2)
            .build()
            .build();

显示执行了step2和step3，但没有执行step4。所以问题是这个流的定义。

您需要在Step3和Flow3之间定义一个并行流，就像您对Flow1和Flow2所做的那样。这是一个示例：

@Bean
public Job Job() {

    Flow flow1 = new FlowBuilder<SimpleFlow>("flow1")
            .start(step1())
            .build();

    Flow flow3 = new FlowBuilder<SimpleFlow>("flow3")
            .start(step4())
            .build();

    Flow parallelFlow = new FlowBuilder<SimpleFlow>("parallelFlow")
            .start(step3())
            .split(new SimpleAsyncTaskExecutor()).add(flow3)
            .build();

    Flow flow2 = new FlowBuilder<SimpleFlow>("flow2")
            .start(step2())
            .next(parallelFlow)
            .build();

    return jobs.get("job")
            .incrementer(new RunIdIncrementer())
            .start(flow1)
            .split(new SimpleAsyncTaskExecutor()).add(flow2)
            .next(step5())
            .end()
            .build();
}

希望这有帮助。