Spring batch重复步骤，以永无止境的循环结束

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

根据我们对Spring批处理的讨论，在任务小包中执行动态生成的步骤，我正在尝试回答有关如何在实际执行作业之前访问jobParameter的问题。

我假设有restcall将执行批处理。一般来说，这需要采取以下步骤。1.接收带有参数的rest调用的一段代码2。创建一个新的springcontext（有一些方法可以重用现有的上下文并再次启动作业，但在重用步骤、读者和作者时会出现一些问题）。开始工作

最简单的解决方案是将从服务接收的jobparameter存储为系统属性，然后在步骤3中构建作业时访问该属性。但如果多个用户同时启动作业，这可能会导致问题。

加载时，还有其他方法可以将参数传递到spring上下文中。但这取决于您设置上下文的方式。例如，如果您直接将SpringBoot用于第2步，您可以编写这样的方法：

private int startJob(Properties jobParamsAsProps) {
  SpringApplication springApp = new SpringApplication(.. my config classes ..);
  springApp.setDefaultProperties(jobParamsAsProps);

  ConfigurableApplicationContext context = springApp.run();
  ExitCodeGenerator exitCodeGen = context.getBean(ExitCodeGenerator.class);
  int code = exitCodeGen.getExitCode();
  context.close();
  return cod;
}

通过这种方式，您可以使用标准值或ConfigurationProperties注释正常访问属性。

Answer 2

从你的问题和代码中，我根据你检索的日期数量（这发生在实际工作开始之前）扣除，你将执行一个步骤来计算你有日期的次数。

我建议更改设计。创建一个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”输入（方法也略有不同）

在类路径的某个地方添加以下示例。属性文件：

sql.statement="select * from awesome"

并将以下注释添加到JobDatesCreator类中

@PropertySource("classpath:example.properties")

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

可以使用特定的命令行开关启动（例如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;
}

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