Spring批处理重复步骤以永无止境的循环结束

Question

我有一个 Spring 批处理工作，我想执行以下操作...

Step 1 - 
   Tasklet - Create a list of dates, store the list of dates in the job execution context.

Step 2 - 
   JDBC Item Reader - Get list of dates from job execution context.
                      Get element(0) in dates list. Use is as input for jdbc query. 
                      Store element(0) date is job execution context 
                      Remove element(0) date from list of dates
                      Store element(0) date in job execution context                 
   Flat File Item Writer - Get element(0) date from job execution context and use for file name.

Then using a job listener repeat step 2 until no remaining dates in the list of dates.

我已经创建了作业，并且它在第一次执行第 2 步时可以正常工作。但第 2 步没有按照我的意愿重复。我知道这一点，因为当我调试我的代码时，它只会在第 2 步的初始运行中中断。

然而，即使我知道它不是，它仍然继续向我提供如下消息，就好像它正在运行第 2 步一样。

2016-08-10 22:20:57.842  INFO 11784 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Duplicate step [readStgDbAndExportMasterListStep] detected in execution of job=[exportMasterListCsv]. If either step fails, both will be executed again on restart.
2016-08-10 22:20:57.846  INFO 11784 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [readStgDbAndExportMasterListStep]

这以永无止境的循环结束。

有人可以帮我弄清楚为什么我的 stpe 2 只运行一次或提出建议吗？

提前致谢

我为我的代码添加了两个指向 PasteBin 的链接，以免污染这篇文章。

http://pastebin.com/QhExNikm (作业配置)

http://pastebin.com/sscKKWRk (通用作业配置)

http://pastebin.com/Nn74zTpS (步骤执行监听器)

Answer 1

从您的问题和您的代码中，我根据您检索的日期数量(这发生在实际工作开始之前)推断出这一点，您将根据日期的次数执行一个步骤。

我建议更改设计。创建一个 java 类，它将为您提供日期作为列表，并根据该列表动态创建您的步骤。像这样的东西:

@EnableBatchProcessing
public class JobConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;  

    @Autowired
    private JobDatesCreator jobDatesCreator;

    @Bean
    public Job executeMyJob() {
        List<Step> steps = new ArrayList<Step>();
        for (String date : jobDatesCreator.getDates()) {
            steps.add(createStep(date));
        }

        return jobBuilderFactory.get("executeMyJob")
                .start(createParallelFlow(steps))
                .end()
                .build();       
    }

    private Step createStep(String date){
        return stepBuilderFactory.get("readStgDbAndExportMasterListStep" + date)
                .chunk(your_chunksize)
                .reader(your_reader)
                .processor(your_processor)
                .writer(your_writer)                                
                .build();       
    }   

    private Flow createParallelFlow(List<Step> steps) {
        SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
        // max multithreading = -1, no multithreading = 1, smart size = steps.size()
        taskExecutor.setConcurrencyLimit(1); 

        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();
    }  
}

编辑:添加了“jobParameter”输入(方法也略有不同)

在类路径的某处添加以下示例 .properties 文件:

sql.statement="select * from awesome"

并将以下注释添加到您的 JobDatesCreator 类

@PropertySource("classpath:example.properties")

您也可以提供特定的 sql 语句作为命令行参数。从 Spring 文档:

you can launch with a specific command line switch (e.g. java -jar app.jar --name="Spring").

有关更多信息，请参阅 http://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-external-config.html

获取日期的类(为什么为此使用 tasklet？):

@PropertySource("classpath:example.properties")
public class JobDatesCreator {

    @Value("${sql.statement}")
    private String sqlStatement;

    @Autowired
    private CommonExportFromStagingDbJobConfig commonJobConfig; 

    private List<String> dates; 

    @PostConstruct
    private void init(){
        // Execute your logic here for getting the data you need.
        JdbcTemplate jdbcTemplate = new JdbcTemplate(commonJobConfig.onlineStagingDb);
        // acces to your sql statement provided in a property file or as a command line argument
        System.out.println("This is the sql statement I provided in my external property: " + sqlStatement);

        // for now..
        dates = new ArrayList<>();
        dates.add("date 1");
        dates.add("date 2");
    }

    public List<String> getDates() {
        return dates;
    }

    public void setDates(List<String> dates) {
        this.dates = dates;
    }
}

我还注意到您有很多可以轻松重构的重复代码。现在对于每个作家，你都有这样的事情:

@Bean
public FlatFileItemWriter<MasterList> division10MasterListFileWriter() {
    FlatFileItemWriter<MasterList> writer = new FlatFileItemWriter<>();
    writer.setResource(new FileSystemResource(new File(outDir, MerchHierarchyConstants.DIVISION_NO_10 )));
    writer.setHeaderCallback(masterListFlatFileHeaderCallback());
    writer.setLineAggregator(masterListFormatterLineAggregator());
    return writer;
}

考虑使用这样的东西:

public FlatFileItemWriter<MasterList> divisionMasterListFileWriter(String divisionNumber) {
    FlatFileItemWriter<MasterList> writer = new FlatFileItemWriter<>();
    writer.setResource(new FileSystemResource(new File(outDir, divisionNumber )));
    writer.setHeaderCallback(masterListFlatFileHeaderCallback());
    writer.setLineAggregator(masterListFormatterLineAggregator());
    return writer;
}

由于并非所有代码都可以正确复制您的问题，因此此答案是解决您问题的建议/指示。