本文详解如何在 Spring Boot 中使用 Spring Batch 的 Split 机制，让步骤 B 和 C 并行执行（多线程），确保步骤 D 仅在二者均成功完成后启动，同时兼容 Quartz 定时调度。

本文详解如何在 spring boot 中使用 spring batch 的 `split` 机制，让步骤 b 和 c 并行执行（多线程），确保步骤 d 仅在二者均成功完成后启动，同时兼容 quartz 定时调度。

在 Spring Batch 中，默认的 next() 链式流程是串行、单线程执行的，无法满足“B 和 C 并行、D 等待两者完成”的拓扑需求。正确解法是引入 Flow.split() + 异步 TaskExecutor，将 B 和 C 封装为两个独立子流（sub-flow），由线程池并发驱动；Spring Batch 内置的 SplitFlow 会自动等待所有分支完成，并聚合 ExitStatus，从而安全触发后续步骤 D。

以下是完整的、可直接运行的 Spring Boot 配置示例（基于 Spring Batch 5.x / Spring Boot 3.x，适配 Jakarta EE 命名空间）：

@Configuration
@EnableBatchProcessing
public class BatchJobConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private DataSource dataSource;

    // ✅ 关键：必须配置异步 TaskExecutor（否则 split 仍串行）
    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);              // 根据 B/C 资源消耗调整
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("batch-parallel-");
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.initialize();
        return executor;
    }

    // 步骤 A：清空表（JDBC）
    @Bean
    public Step stepA() {
        return stepBuilderFactory.get("stepA")
                .tasklet((contribution, chunkContext) -> {
                    JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
                    jdbcTemplate.update("TRUNCATE TABLE target_table");
                    return RepeatStatus.FINISHED;
                }, transactionManager())
                .build();
    }

    // 步骤 B：批量写入（JDBC Chunk）
    @Bean
    public Step stepB() {
        return stepBuilderFactory.get("stepB")
                .<String, String>chunk(2000, transactionManager())
                .reader(jdbcCursorItemReaderB())  // 自定义 reader，例如查 source_table_b
                .writer(jdbcBatchItemWriterB())
                .build();
    }

    // 步骤 C：批量写入（另一数据源或表）
    @Bean
    public Step stepC() {
        return stepBuilderFactory.get("stepC")
                .<String, String>chunk(2000, transactionManager())
                .reader(jdbcCursorItemReaderC())
                .writer(jdbcBatchItemWriterC())
                .build();
    }

    // 步骤 D：PL/SQL 存储过程调用（依赖 B & C 结果）
    @Bean
    public Step stepD() {
        return stepBuilderFactory.get("stepD")
                .tasklet((contribution, chunkContext) -> {
                    JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
                    jdbcTemplate.update("CALL process_aggregated_data()");
                    return RepeatStatus.FINISHED;
                }, transactionManager())
                .build();
    }

    // ✅ 构建并行子流：flowB 和 flowC 同时执行
    @Bean
    public Flow parallelBCFlow() {
        return new FlowBuilder<SimpleFlow>("parallelBCFlow")
                .split(taskExecutor())  // ⚠️ 必须传入异步 TaskExecutor
                .add(
                        new FlowBuilder<SimpleFlow>("flowB").start(stepB()).build(),
                        new FlowBuilder<SimpleFlow>("flowC").start(stepC()).build()
                )
                .build();
    }

    // 主 Job：A → (B||C) → D
    @Bean
    public Job demoJob() {
        return jobBuilderFactory.get("demoJob")
                .listener(jobExecutionListener())
                .start(stepA())
                .next(parallelBCFlow())  // ← 替代原来的 .next(B()).next(C())
                .next(stepD())
                .end()
                .build();
    }

    // 其他必要 Bean（省略事务管理器、监听器等，按需补充）
    @Bean
    public PlatformTransactionManager transactionManager() {
        return new DataSourceTransactionManager(dataSource);
    }

    @Bean
    public JobExecutionListener jobExecutionListener() {
        return new JobExecutionListener() {
            @Override
            public void beforeJob(JobExecution jobExecution) {
                System.out.println("▶ Job started: " + jobExecution.getJobInstance().getJobName());
            }
            @Override
            public void afterJob(JobExecution jobExecution) {
                System.out.println("✔ Job completed with status: " + jobExecution.getStatus());
            }
        };
    }

    // 示例 Reader/Writers（简化版，实际需配置 SQL 和映射）
    @Bean
    public JdbcCursorItemReader<String> jdbcCursorItemReaderB() {
        JdbcCursorItemReader<String> reader = new JdbcCursorItemReader<>();
        reader.setDataSource(dataSource);
        reader.setSql("SELECT id FROM source_b WHERE processed = false");
        reader.setRowMapper((rs, i) -> rs.getString("id"));
        return reader;
    }

    @Bean
    public JdbcBatchItemWriter<String> jdbcBatchItemWriterB() {
        JdbcBatchItemWriter<String> writer = new JdbcBatchItemWriter<>();
        writer.setDataSource(dataSource);
        writer.setSql("INSERT INTO target_table (id, source) VALUES (?, 'B')");
        writer.setItemPreparedStatementSetter((item, ps) -> ps.setString(1, item));
        return writer;
    }

    // 类似定义 jdbcCursorItemReaderC / jdbcBatchItemWriterC ...
}

✅ 关键注意事项：

• TaskExecutor 是并行前提：SplitFlow 默认使用 SyncTaskExecutor（即无并发），必须显式注入 ThreadPoolTaskExecutor 或 SimpleAsyncTaskExecutor；推荐前者以复用线程、避免资源耗尽。
• 事务隔离：B 和 C 是独立事务（每个 Step 拥有自己 TransactionManager），互不影响；但需确保它们写入的数据对步骤 D 可见（如通过 READ_COMMITTED 隔离级别或显式 commit 后刷新）。
• Quartz 集成无需额外改动：只要 Quartz 触发的是同一个 JobLauncher 执行 demoJob，并行逻辑完全由 Spring Batch 内部调度，与 Quartz 的单线程 Worker 无关——因为并行发生在 Job 执行内部（即 Quartz 启动一个 Job 实例，该实例内部开启多线程）。
• 错误处理：若 B 或 C 失败，SplitFlow 会中断整个 Job，stepD() 不会执行；可通过 onFailure() 配置重试或跳过策略（需结合 FaultTolerantStepBuilder）。
• 监控与日志：建议启用 Spring Batch 的 JobRepository（默认内存或数据库持久化），配合 Actuator /actuator/batch 端点观察各步骤并发状态。

通过以上配置，即可精准实现 A → (B∥C) → D 的执行图谱：步骤 A 单线程完成；B 和 C 在独立线程中并行执行（耗时长者决定整体分支耗时）；步骤 D 在二者全部成功提交后才启动，保障业务强一致性。