java - 为什么在Spring Batch中不调用拆分流？

我有以下作业配置：

@Bean
public Job job(Step databaseToDataBaseLowercaseSlaveStep) {
    return jobBuilderFactory.get("myJob")
            .incrementer(new RunIdIncrementer())
            .flow(csvToDbLowercaseStep())
            .next(databaseToDataBaseLowercaseSlaveStep)
            .split(jobTaskExecutor())
            .add(new FlowBuilder<Flow>("flow2")
                    .start(notificationStep())
                    .build()
            )
            .end()
            .build();
}

预期的动作顺序：


执行csvToDbLowercaseStep
并行运行2个步骤
a）databaseToDataBaseLowercaseSlaveStep
b）notificationStep


实际动作顺序：


执行csvToDbLowercaseStep
执行databaseToDataBaseLowercaseSlaveStep


因此，根本没有调用notificationStep。为什么？我该如何纠正？

更新资料

@Bean
public TaskExecutor jobTaskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    // there are 21 sites currently hence we have 21 threads
    taskExecutor.setMaxPoolSize(30);
    taskExecutor.setCorePoolSize(25);
    taskExecutor.setThreadGroupName("cust-job-exec-");
    taskExecutor.setThreadNamePrefix("cust-job-exec-");
    taskExecutor.afterPropertiesSet();
    return taskExecutor;
}

最佳答案

不支持将拆分添加到状态链，这是执行所需操作的正确方法（同步任务1，然后并行执行任务2和3）：

    public Job job() {

        final Flow masterFlow = new FlowBuilder<Flow>("flow1").start(step("step1")).build();

        final Flow slaveFlow = new FlowBuilder<Flow>("flow2").split(new SimpleAsyncTaskExecutor())
                .add(
                        new FlowBuilder<Flow>("flow2.1").start(step("step2.1")).build(),
                        new FlowBuilder<Flow>("flow2.2").start(step("step2.2")).build())
                .build();

        return (jobBuilderFactory
                .get("job")
                .incrementer(new RunIdIncrementer())
                .start(masterFlow)
                .next(slaveFlow)
                .build())
                        .build();
    }

    private TaskletStep step(final String name) {

        return stepBuilderFactory.get(name)
                .tasklet((StepContribution contribution, ChunkContext chunkContext) -> {

                    System.out.println(name + " start");
                    Thread.sleep(1000);
                    System.out.println(name + " end");

                    return RepeatStatus.FINISHED;
                })
                .build();
    }

步骤1开始

第一步

step2.1开始

step2.2开始

step2.1结束

step2.2结束

希望这可以帮助。

更新资料

您的代码正在尝试将拆分添加到状态链中，并且根据FlowBuilder.SplitBuilder的文档，它根本不受支持。

     * <em>Note:</em> Adding a split to a chain of states is not supported.  For example, the following configuration
     * is not supported.  Instead, the configuration would need to create a flow3 that was the split flow and assemble
     * them separately.
     *
     * <pre>
     * // instead of this
     * Flow complexFlow = new FlowBuilder&lt;SimpleFlow&gt;("ComplexParallelFlow")
     *                       .start(flow1)
     *                       .next(flow2)
     *                       .split(new SimpleAsyncTaskExecutor())
     *                       .add(flow3, flow4)
     *                       .build();
     *
     * // do this
     * Flow splitFlow = new FlowBuilder&lt;SimpleFlow&gt;("parallelFlow")
     *                       .start(flow3)
     *                       .split(new SimpleAsyncTaskExecutor())
     *                       .add(flow4).build();
     *
     * Flow complexFlow = new FlowBuilder&lt;SimpleFlow&gt;("ComplexParallelFlow")
     *                       .start(flow1)
     *                       .next(flow2)
     *                       .next(splitFlow)
     *                       .build();
     * </pre>