嘿，这个需求我之前做类似批量作业时刚好碰到过，用JobExecutionDecider来实现动态分支完全能满足你按需选择输出的需求，我给你拆解下具体实现步骤和代码示例，应该能帮你理清疑问～

核心思路 #

把读取数据库的逻辑做成独立公共Step，再通过**决策器（Decider）**读取传入的output参数，动态决定后续执行「写入CSV」或「写入Topic」的Step。不用再手动注释切换代码，完全靠参数控制，灵活度拉满。

具体实现步骤 #

• 1. 定义两个输出Writer Bean #

先把你已经验证过的CSV Writer和Topic Writer都注册成Spring Bean，确保各自功能正常：

// CSV Writer
@Bean
@StepScope
public FlatFileItemWriter<YourDataModel> csvWriter() {
    FlatFileItemWriter<YourDataModel> writer = new FlatFileItemWriter<>();
    writer.setResource(new FileSystemResource("output.csv"));
    writer.setLineAggregator(new DelimitedLineAggregator<>() {{
        setDelimiter(",");
        setFieldExtractor(new BeanWrapperFieldExtractor<>() {{
            setNames(new String[]{"id", "name", "value"}); // 替换成你的业务字段
        }});
    }});
    return writer;
}

// Topic Writer
@Bean
@StepScope
public KafkaItemWriter<YourDataModel> topicWriter(KafkaTemplate<String, YourDataModel> kafkaTemplate) {
    KafkaItemWriter<YourDataModel> writer = new KafkaItemWriter<>();
    writer.setKafkaTemplate(kafkaTemplate);
    writer.setTopic("your-target-topic");
    return writer;
}

• 2. 实现自定义JobExecutionDecider #

这个决策器的核心就是读取output参数，判断走哪个分支：

@Component
public class OutputTypeDecider implements JobExecutionDecider {

    @Override
    public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
        String outputType = jobExecution.getJobParameters().getString("output");
        
        if ("csv".equalsIgnoreCase(outputType)) {
            return new FlowExecutionStatus("CSV");
        } else if ("topic".equalsIgnoreCase(outputType)) {
            return new FlowExecutionStatus("TOPIC");
        } else {
            // 参数不合法时返回失败，也可以设置默认分支
            return FlowExecutionStatus.FAILED;
        }
    }
}

• 3. 配置带分支的Job流程 #

把公共读取Step、两个输出Step和决策器串起来，构建动态分支的Job：

@Configuration
public class BatchJobConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    private OutputTypeDecider outputTypeDecider;
    @Autowired
    private ItemReader<YourDataModel> dbReader; // 你已有的数据库Reader
    @Autowired
    private FlatFileItemWriter<YourDataModel> csvWriter;
    @Autowired
    private KafkaItemWriter<YourDataModel> topicWriter;

    // 公共读取数据Step
    @Bean
    public Step readDbStep() {
        return stepBuilderFactory.get("readDbStep")
                .<YourDataModel, YourDataModel>chunk(100)
                .reader(dbReader)
                .build();
    }

    // 写入CSV的Step
    @Bean
    public Step writeCsvStep() {
        return stepBuilderFactory.get("writeCsvStep")
                .<YourDataModel, YourDataModel>chunk(100)
                .writer(csvWriter)
                .build();
    }

    // 写入Topic的Step
    @Bean
    public Step writeTopicStep() {
        return stepBuilderFactory.get("writeTopicStep")
                .<YourDataModel, YourDataModel>chunk(100)
                .writer(topicWriter)
                .build();
    }

    // 构建动态分支Job
    @Bean
    public Job dynamicOutputJob() {
        return jobBuilderFactory.get("dynamicOutputJob")
                .start(readDbStep())
                .next(outputTypeDecider)
                .on("CSV").to(writeCsvStep())
                .on("TOPIC").to(writeTopicStep())
                .on("FAILED").fail()
                .end()
                .build();
    }
}

• 4. 传入参数启动作业 #

现在只需要在启动时传入output参数，就能指定输出方式：

• 命令行启动Spring Boot应用：

java -jar your-batch-app.jar --output=csv
# 或者指定输出到Topic
java -jar your-batch-app.jar --output=topic

• 代码内触发作业：

@Autowired
private JobLauncher jobLauncher;
@Autowired
private Job dynamicOutputJob;

public void runBatchJob(String outputType) throws Exception {
    JobParameters jobParameters = new JobParametersBuilder()
            .addString("output", outputType)
            .addLong("timestamp", System.currentTimeMillis()) // 避免重复执行同一参数的作业
            .toJobParameters();
    jobLauncher.run(dynamicOutputJob, jobParameters);
}

额外注意点 #

• 给Writer加上@StepScope，方便后续如果需要从参数动态读取输出路径、Topic名称时，用@Value("#{jobParameters['xxx']}")注入。
• 决策器里的参数判断忽略大小写，提升容错性。
• 如果需要默认分支，可以在decider里当参数不存在时返回new FlowExecutionStatus("DEFAULT")，然后在Job配置里追加.on("DEFAULT").to(writeCsvStep())。

这样配置完，业务方要什么输出方式传对应参数就行，完全不用手动改代码啦～

内容的提问来源于stack exchange，提问作者CardsFan