当达到下一个步骤的条件时如何无限地返回到上一个步骤 Spring Batch

Question

你好 Spring Batch 社区！我有一个带有标题和正文的输入平面文件。标题是 1 行（自然..）和 5 个参数。 Body 可以达到多达 100 万条记录，每条记录有 12 个参数。

输入文件：

01.01.2017|SUBDCOBR|12:21:23|01/12/2016|31/12/2016
01.01.2017|12345678231234|0002342434|BORGIA RUBEN|27-32548987-9|FA|A|2062-
00010443/444/445|142,12|30/08/2017|142,01
01.01.2017|12345673201234|2342434|ALVAREZ ESTHER|27-32533987-9|FA|A|2062-
00010443/444/445|142,12|30/08/2017|142,02
01.01.2017|12345673201234|0002342434|LOPEZ LUCRECIA|27-32553387-9|FA|A|2062-
00010443/444/445|142,12|30/08/2017|142,12
01.01.2017|12345672301234|0002342434|SILVA JESUS|27-32558657-9|NC|A|2062-
00010443|142,12|30/08/2017|142,12
.
.   
.

我需要将它写入一个具有特定格式的 .txt 文件，并采用这种特定的结构：

HEADER (8 customed lines, using data from HEADER input)
TITLE OF COLUMNS (1 line)
DETAILS (17 records from the body)
line break
SAME HEADER
SAME TITLE OF COLUMNS
DETAILS (next 17 records from the body)
line break
...
...  
...
REPEAT until end of file

我所做的是...创建一个 stepHeader 和一个 stepBody 。他们每个人都有自己的阅读器、处理器（业务格式化程序）和编写器。

这项工作将只有这两个简单的步骤。

@Bean
public Job job() throws Exception {
        return jobBuilderFactory.get("job")
                .incrementer(new RunIdIncrementer())
                .listener(new JobListener())
                .start(stepHeader())
                .next(stepBody())
                .on("BACK TO STEPHEADER").to(stepHeader())
                .on("END").end().build()
                .build();
}

我读取的header配置MaxItemCount=1，并映射到CabeceraFacturacion：

    @Bean
    public FlatFileItemReader<CabeceraFacturacion> readerCabecera() throws Exception{

    FlatFileItemReader<CabeceraFacturacion> reader = new FlatFileItemReader<>();

    reader.setLinesToSkip(0);
    reader.setMaxItemCount(1);
    reader.setResource(new ClassPathResource("/inputFiles/input.txt"));

    DefaultLineMapper<CabeceraFacturacion> cabeceraLineMapper = new DefaultLineMapper<>();

    DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer("|"); // en el default, la coma es el separador
    tokenizer.setNames(new String[] {"printDate", "reportIdentifier", "tituloReporte", "fechaDesde", "fechaHasta"});

    cabeceraLineMapper.setLineTokenizer(tokenizer);
    cabeceraLineMapper.setFieldSetMapper(new CabeceraFieldSetMapper());
    cabeceraLineMapper.afterPropertiesSet();

    reader.setLineMapper(cabeceraLineMapper);

    return reader;
}

我这样读的正文，跳过第一行，并将其映射到 DetalleFacturacion：

@Bean
    public FlatFileItemReader<DetalleFacturacion> readerDetalleFacturacion(){

    FlatFileItemReader<DetalleFacturacion> reader = new FlatFileItemReader<>();

    reader.setLinesToSkip(1);
    //reader.setMaxItemCount(17);
    reader.setResource(new ClassPathResource("/inputFiles/input.txt"));

    DefaultLineMapper<DetalleFacturacion> detalleLineMapper = new DefaultLineMapper<>();

    DelimitedLineTokenizer tokenizerDet = new DelimitedLineTokenizer("|"); // en el default, la coma es el separador
    tokenizerDet.setNames(new String[] {"fechaEmision", "tipoDocumento", "letra", "nroComprobante",  
                                "nroCliente", "razonSocial", "cuit", "montoNetoGP", "montoNetoG3",
                                "montoExento", "impuestos", "montoTotal"});



    detalleLineMapper.setLineTokenizer(tokenizerDet);
    detalleLineMapper.setFieldSetMapper(new DetalleFieldSetMapper());
    detalleLineMapper.afterPropertiesSet();
    reader.setLineMapper(detalleLineMapper);


    return reader;
}   

我的步骤：

@Bean
public Step stepHeader() throws Exception {
    return stepBuilderFactory.get("stepHeader")
            .<CabeceraFacturacion, CabeceraFacturacion> chunk(17)
            .faultTolerant()
            .listener(new ChunkListener())
            .reader(readerCabecera())
            .writer(writerCabeceraFact())
            .allowStartIfComplete(true)
            .build();
}


@Bean
public Step stepBody() {
    return stepBuilderFactory.get("stepBody")
            .<DetalleFacturacion, DetalleFacturacion> chunk(17)
            .chunk(17)
            .faultTolerant()
            .listener(new ChunkListener())
            .reader(readerDetalleFacturacion())
            .writer(writerDetalleFact())
            .listener(new StepExecutionListener() {

                @Override
                public ExitStatus afterStep(StepExecution stepExecution) {
                    if(stepExecution.getWriteCount()==17) {

                        return new ExitStatus("BACK TO STEPHEADER");                    
                    };

//                      if(stepExecution.getReadCount()<17) {
//                          return new ExitStatus("END");                   

//                      }

                    return null;
                }

                @Override
                public void beforeStep(StepExecution stepExecution) {


                }


            })
            .allowStartIfComplete(true)
            .build();
}

1) 我不知道如何在文件结束之前无限地返回到 StepHeader。在那里我尝试使用 stepExecution.getWriteCount(17).. 但我不确定这是不是这样。

2) 我不知道每次循环时如何读取 17 条不同的记录（我设法让它循环，但它会一遍又一遍地写入相同的前 17 条记录，直到我手动停止工作。我现在知道了在 Spring Batch 过程中不建议使用循环。

3) 如果有人对实现我的目标的其他方式有任何想法，将不胜欢迎。

4) 有没有办法让决策者一直“听到”，如果满足某些条件，则发送命令打印标题或正文？

到目前为止，我达到的最大值是只读取和写入一次标题...并在下一步读取和写入正文的 17 行。

谢谢大家！ 干杯！！

Answer 1

不确定我是否正确理解了您的问题，但这就是您想要实现的目标

第 1 步：从文件中读取标题

第2步：读取文件，处理数据并写入某个文件直到某个条件A

第 3 步：在条件 A 上转到第 1 步

可以有多个选项来配置它。我能想到的一个是为流程决策添加额外的步骤。下面是示例配置。

注意我没有测试过，你可能需要做一些修改

@Bean
    public Job conditionalJob(JobBuilderFactory jobs, Step conditionalStep1, Step conditionalStep2, Step conditionalStep3, Step conditionalStep4, Step conditionalStep5) throws Exception {
        return jobs.get("conditionalJob")
                .incrementer(new RunIdIncrementer())
                .flow(flowDesider).on("HEADER").to(step1).next("flowDesider")
                .from(flowDesider).on("BODAY").to(step2).next("flowDesider")
                .from(flowDesider).on("*").stop()
                .end()
                .build();
    }



public class flowDesider implements Tasklet{

private Logger logger = LoggerFactory.getLogger(this.getClass());

@Override
public RepeatStatus execute(StepContribution contribution,
        ChunkContext chunkContext) throws Exception {
    logger.info("flowDesider");

    //put your flow logic header 
            //you can use step excequation to pass infomrtion from one step to onother      
    if(codition1)
        return status as HEADER
    if (condition2)
        return status as Body
    if condition3
        return status as complited 


}