Spring Batch如何在job运行时基于inspect步骤结果动态添加Step？

方案1：基于Partitioner的分区实现（推荐） #

这个是Spring Batch原生支持的运行时动态拆分任务的方案，完全满足你需要的动态步骤数量、单文件失败隔离、不同csv用不同Reader的需求，无需预定义步骤：

• 第一步：先执行两个固定步骤：下载Zip压缩包、Inspect压缩包内容。Inspect步骤执行完成后，将识别到的所有csv文件的元信息（压缩包内路径、csv类型标识、对应Reader的配置参数等）存入JobExecutionContext，仅存可序列化的元数据即可，不要存InputStream对象避免序列化失败。
• 第二步：定义第三个步骤为分区主步骤，自定义实现Partitioner接口：
  重写partition方法时从JobExecutionContext读取之前存入的csv元信息列表，每个csv对应生成一个独立分区，将该csv的元信息存入分区的ExecutionContext中。
• 第三步：配置分区从步骤为@StepScope作用域，运行时会根据分区数量动态生成对应数量的从步骤实例：
  每个从步骤执行时，从当前分区的ExecutionContext拿到对应csv的元信息，动态实例化对应类型的FlatFileItemReader，直接通过ZipFileSystem读取压缩包内对应条目的InputStream，无需解压到磁盘，写入逻辑复用已有的JdbcBatchItemWriter即可。
• 优势：天然支持单文件失败隔离，单个分区（对应单个csv）执行失败默认不会终止其他分区的执行；支持配置异步任务执行器实现多csv并行处理；所有步骤的执行日志、监控数据天然区分，支持单独重启失败的csv处理任务。

示例代码片段：

// 自定义Partitioner
public class CsvZipPartitioner implements Partitioner {
    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {
        // 从JobExecutionContext获取inspect步骤存入的csv元信息列表
        List<CsvMeta> csvMetaList = ...;
        Map<String, ExecutionContext> partitions = new HashMap<>();
        for (int i = 0; i < csvMetaList.size(); i++) {
            ExecutionContext context = new ExecutionContext();
            context.put("csvMeta", csvMetaList.get(i));
            partitions.put("csvPartition-" + i, context);
        }
        return partitions;
    }
}

// 分区从步骤配置
@Bean
@StepScope
public Step csvProcessStep(StepBuilderFactory stepBuilderFactory,
                           @Value("#{stepExecutionContext['csvMeta']}") CsvMeta csvMeta) {
    // 动态根据csv类型创建对应Reader
    FlatFileItemReader<?> reader = createReaderByCsvType(csvMeta);
    return stepBuilderFactory.get("csvProcessStep-" + csvMeta.getPath())
            .chunk(1000)
            .reader(reader)
            .writer(jdbcBatchItemWriter())
            .build();
}

方案2：基于JobExecutionDecider的循环执行实现 #

如果不需要并行处理，也可以用判断器实现循环动态调用处理步骤：

• 前两个固定步骤执行完成后，将csv元信息列表和当前处理下标（初始为0）存入JobExecutionContext
• 自定义JobExecutionDecider，每次执行时判断当前下标是否小于csv总数量：
  如果小于，返回"PROCESS_CSV"状态，触发csv处理步骤执行；如果大于等于，返回"COMPLETED"结束作业。
• csv处理步骤同样配置为@StepScope，每次执行时根据当前下标读取对应csv的元信息，初始化对应Reader执行处理，步骤执行完成后将下标+1，重新回到Decider做下一轮判断。
• 注意：处理步骤名称要动态带上csv标识，方便日志排查问题；默认串行执行，需要并行的话要自行调整实现。

注意事项 #

• 不要将ZipInputStream等不可序列化的对象存入ExecutionContext，仅存元数据即可，处理步骤执行时重新打开Zip文件读取对应条目即可，读取单个条目不会占用过多磁盘空间。
• 可以给单个csv处理步骤配置独立的重试、跳过策略，进一步隔离单文件的处理风险。

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