Spring批处理条件流创建无限循环

问题描述 投票:0回答:1

我有一个简单的三步流程:

public Job myJob() {
    Step extract = extractorStep();
    Step process = filesProcessStep();
    Step cleanup = cleanupStep();

    return jobBuilderFactory.get("my-job")
          .flow(echo("Starting job"))
          .next(extract)
          .next(process)
          .next(cleanup)
          .next(echo("Ending job"))
          .end()
          .build();
  }

现在我想使用StepExecutionListener.afterStep()中的ExitStatus添加错误处理。任何错误都应该转发到清理步骤。所以我改为下面的代码:

 public Job myJob() {
    Step extract = extractorStep();
    Step process = filesProcessStep();
    Step cleanup = cleanupStep();

    return jobBuilderFactory.get("my-job")
          .start(echo("Starting batch job"))

          .next(extract).on(ExitStatus.FAILED.getExitCode()).to(cleanup)
          .from(extract).on("*").to(process)

          .next(process).on(ExitStatus.FAILED.getExitCode()).to(cleanup)
          .from(process).on("*").to(cleanup)

          .next(echo("End batch job"))
          .end()
          .build();
  }

现在我有一个无限循环清理步骤。我想帮助纠正这个流程。

spring-batch spring-java-config
1个回答
2
投票

在您的示例中,从cleanup开始,流程未定义。你应该准确地说,从cleanup流量必须继续使用echo .from(cleanup).to(echo("End batch job")).end()。这是一个例子:

import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableBatchProcessing
public class MyJob {

    @Autowired
    private JobBuilderFactory jobs;

    @Autowired
    private StepBuilderFactory steps;

    @Bean
    public Step extractorStep() {
        return steps.get("extractorStep")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("extractorStep");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Step filesProcessStep() {
        return steps.get("filesProcessStep")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("filesProcessStep");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Step cleanupStep() {
        return steps.get("cleanupStep")
                .tasklet((contribution, chunkContext) -> {
                    System.out.println("cleanupStep");
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    public Step echo(String message) {
        return steps.get("echo-" + message)
                .tasklet((contribution, chunkContext) -> {
                    System.out.println(message);
                    return RepeatStatus.FINISHED;
                })
                .build();
    }

    @Bean
    public Job job() {
        Step start = echo("Starting batch job");
        Step extract = extractorStep();
        Step process = filesProcessStep();
        Step cleanup = cleanupStep();
        Step stop = echo("End batch job");

        return jobs.get("job")
                .start(start).on("*").to(extract)

                    .from(extract).on(ExitStatus.FAILED.getExitCode()).to(cleanup)
                    .from(extract).on("*").to(process)

                    .from(process).on("*").to(cleanup)

                    .from(cleanup).next(stop)
                    .build()

                .build();
    }

    public static void main(String[] args) throws Exception {
        ApplicationContext context = new AnnotationConfigApplicationContext(MyJob.class);
        JobLauncher jobLauncher = context.getBean(JobLauncher.class);
        Job job = context.getBean(Job.class);
        jobLauncher.run(job, new JobParameters());
    }

}

它打印:

Starting batch job
extractorStep
filesProcessStep
cleanupStep
End batch job

如果extractorStep失败,例如:

@Bean
public Step extractorStep() {
    return steps.get("extractorStep")
            .tasklet((contribution, chunkContext) -> {
                System.out.println("extractorStep");
                chunkContext.getStepContext().getStepExecution().setExitStatus(ExitStatus.FAILED);
                return RepeatStatus.FINISHED;
            })
            .build();
}

流将跳过filesProcessStep并去清理:

Starting batch job
extractorStep
cleanupStep
End batch job

希望这可以帮助。

© www.soinside.com 2019 - 2024. All rights reserved.