本文详解如何在 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 ...
}</string></string></string></string></simpleflow></simpleflow></simpleflow></string></string>
✅ 关键注意事项：
-
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 在二者全部成功提交后才启动，保障业务强一致性。
