如何在单个作业中执行多个弹簧批处理步骤

Question

@Qualifier("billingJob")
@Bean
public Job billingJob(@Qualifier("fetchAgreementTasklet")Step fetchAgreementTasklet,
@Qualifier("fetchAgreementRecurringItemsTasklet") Step fetchAgreementRecurringItemsTasklet,
@Qualifier("fetchItemsHistoryTasklet") Step fetchItemsHistoryTasklet ,
@Qualifier("populateAgreementStep") Step populateAgreementStep,
@Qualifier("populateRecurringItemStep") Step populateRecurringItemStep ,
@Qualifier("populateRecurringItemHistoryStep") Step populateRecurringItemHistoryStep ) {

  return jobsBuilderFactory.get("billingJob")                                
           .incrementer(new RunIdIncrementer())
           .flow(fetchAgreementTasklet)
           .next(fetchAgreementRecurringItemsTasklet)
           .next(fetchItemsHistoryTasklet)

           .next(populateAgreementStep).next(populateRecurringItemStep)
           .next(populateRecurringItemHistoryStep)
           .end().build();
}

    @Qualifier("populateRecurringItemStep")
    @Bean
    public Step populateRecurringItemStep(
    @Qualifier("recurringItemReader") ItemReader<RecurringItemRaw> recurringItemReader,
    @Qualifier("recurringItemProcessor") ItemProcessor<RecurringItemRaw, RecurringItem> recurringItemProcessor,
    @Qualifier("recurringItemWriter") ItemWriter<RecurringItem> recurringItemWriter) {
       return stepBuilderFactory.get("populateRecurringItemStep")
        .<RecurringItemRaw, RecurringItem> chunk(10)
        .reader(recurringItemReader).processor(recurringItemProcessor)
        .writer(recurringItemWriter).build();
    } 

@Qualifier("recurringItemReader")
@Bean
public ItemReader<RecurringItemRaw> recurringItemReader() {
FlatFileItemReader<RecurringItemRaw> reader = new FlatFileItemReader<RecurringItemRaw>();
String file = salesforceConfiguration       .getFileLocation(SalesforceConfiguration.TYPE_AGREEMENT_ITEMS);
    LOG.info("::::::: Reading RecurringItem File ::::::: " + file);
    reader.setResource(new FileSystemResource(file));
    reader.setLinesToSkip(1);
    reader.setLineMapper(new DefaultLineMapper<RecurringItemRaw>() {
        {
            setLineTokenizer(new DelimitedLineTokenizer() {
                {
                    setNames(new String[] { "id", "name", "agreementId",
                            "cost", "quantity" });
                }
            });
            setFieldSetMapper(new BeanWrapperFieldSetMapper<RecurringItemRaw>() {
                {
                    setTargetType(RecurringItemRaw.class);
                }
            });
        }
    });
    return reader;
}

@Qualifier("recurringItemProcessor")
@Bean
public ItemProcessor<RecurringItemRaw, RecurringItem> recurringItemprocessor() {
return new RecurringItemProcessor();
}

@Qualifier("recurringItemWriter")
@Bean
public ItemWriter<RecurringItem> recurringItemWriter(DataSource dataSource) {
JdbcBatchItemWriter<RecurringItem> writer = new JdbcBatchItemWriter<RecurringItem>();
writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<RecurringItem>());
writer.setSql(PostgresqlDBQuery.INSERT_RECURRING_ITEM_QRY);
writer.setDataSource(dataSource);
return writer;
}

recurringItemprocessor（）用于将字符串值转换为其他格式

recurringItemWriter（）用于将记录插入到我的postgresql数据库中

recurringItemReader（）用于读取csv文件。

---

这是处理器类公共类RecurringItemProcessor实现ItemProcessor {

private static final Logger LOG = Logger.getLogger(RecurringItemProcessor.class.getName());

@Override
public RecurringItem process(final RecurringItemRaw recurringItemRaw) throws Exception {
    LOG.info("Processing recurring item");
    final RecurringItem item = new RecurringItem(); 
    item.setId(recurringItemRaw.getId());
    item.setName(recurringItemRaw.getName());
    item.setAgreementId(recurringItemRaw.getAgreementId());
    if (recurringItemRaw.getCost().trim().isEmpty()){
        item.setCost(BigDecimal.ZERO);
    }
    else{
        item.setCost(new BigDecimal(recurringItemRaw.getCost())); 
    }
    if (recurringItemRaw.getQuantity().trim().isEmpty()){
        item.setQuantity(0);
    }else{
        item.setQuantity(new BigDecimal(recurringItemRaw.getQuantity()).intValue());
    }
    LOG.info(item.toString());
    return item;
}

}

我遇到了“populateRecurringItemStep”的问题，其中此步骤在endloop中运行。如何解决这个问题..我正在使用spring boot no xml配置..

Answer 1

而不是使用flow()，使用start()，如下所示：

return jobsBuilderFactory.get("billingJob")                                
           .incrementer(new RunIdIncrementer())
           .start(fetchAgreementTasklet)
           .next(fetchAgreementRecurringItemsTasklet)
           .next(fetchItemsHistoryTasklet)

           .next(populateAgreementStep).next(populateRecurringItemStep)
           .next(populateRecurringItemHistoryStep)
           .end().build();
}