基于作业参数配置小枝批写入器

Question

我使用spring引导创建了一个简单的spring批处理作业，它从数据库读取并写入主题。我还有一个钩子，可以在开发过程中注释掉topicWriter并写入csv文件。这两种方法都是通过注释一个输出和运行另一个作者来完成的。(topicWriter或作者)企业现在想要能够管理临时、专题或作家。因此，我选择传递一个包含主题或csv的输出参数。看上去我可以用一个决定器，但这可能是错的。正如下面的代码所显示的那样，代码会抱怨重复的步骤，并且当我试图运行时正在循环。我无法知道如何在没有开始步骤的情况下运行，所以我创建了一个do nothing，因为作业需要在决定器之前先开始一步。所以我想我搞砸了。对解决方案或方向有什么想法吗？

   @Bean
public Job job(@Qualifier("step") Step step) {
    return jobBuilderFactory.get(BatchConstants.JOB_NAME).listener(jobListener())
            .start(step).next(decider()).on("COMPLETED").to(step1(null,null)).from(decider()).on("FAILED").to(step2(null,null)).end().build();
}


  @Bean
protected Step step() {
    return stepBuilderFactory.get("step")
            .tasklet(new Tasklet() {
                public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {
                    return RepeatStatus.FINISHED;

                }
            })
            .build();
}

@Bean
protected Step step1(ItemReader<someDto> reader,
                    ItemWriter<someDto> topicWriter) {
    return stepBuilderFactory.get(BatchConstants.STEP_NAME)
            .<someDto, someDto> chunk(BatchConstants.CHUNKSIZE)
            .reader(reader)
            .writer(topicWriter) // write to kafka topic.
            .build();
}


@Bean
protected Step step2(ItemReader<someDto> reader,
                    ItemWriter<someDto> writer) {
    return stepBuilderFactory.get(BatchConstants.STEP_NAME)
            .<someDto, someDto> chunk(BatchConstants.CHUNKSIZE)
            .reader(reader)
            .writer(writer) // writes to csv
            .build();
}

Answer 1

在Singel步骤中，您可以使用@StepScpoe来定义这个步骤。根据职务参数，您可以选择写入器。

@Bean
    @StepScope
    protected Step step2(ItemReader<someDto> reader,
                        ItemWriter<someDto> writer ,ItemWriter<someDto> topicWriter,"#{jobParameters['writerType']}") final String type ) {

                            ItemWriter<someDto> myWriter;           
                            if(type.equals("topic"))
                            {
                                myWriter=topicWriter;
                            }
                            else
                            {
                                myWriter=writer;
                            }

        return stepBuilderFactory.get(BatchConstants.STEP_NAME)
                .<someDto, someDto> chunk(BatchConstants.CHUNKSIZE)
                .reader(reader)
                .writer(myWriter) 
                .build();
    }