spring-batch

Spring Batch

Spring Boot 3.x 随附 Spring Batch 5。API 相较 4.x 有显著变化——网上大多数示例都是错的。最容易导致 Agent 生成的代码出错的两条规则：

1. 不要添加 @EnableBatchProcessing。 Boot 会自动配置 JobRepository、 JobLauncher 和事务管理器。添加 @EnableBatchProcessing 会禁用该自动配置， 你将失去所有已接入的 Bean。
2. JobBuilderFactory 和 StepBuilderFactory 已被移除。 使用 new JobBuilder(name, jobRepository) / new StepBuilder(name, jobRepository) 构建作业和步骤，并注入 Boot 已创建的 JobRepository 和 PlatformTransactionManager。

作业与步骤（Spring Batch 5 API）

@Configuration
@RequiredArgsConstructor
public class OrderExportJobConfig {

    @Bean
    public Job orderExportJob(JobRepository jobRepository, Step exportStep) {
        return new JobBuilder("orderExportJob", jobRepository)
            .incrementer(new RunIdIncrementer()) // lets the same job be re-run; see "Idempotency"
            .start(exportStep)
            .build();
    }

    @Bean
    public Step exportStep(JobRepository jobRepository,
                           PlatformTransactionManager txManager, // Boot's, injected — do NOT new one up
                           ItemReader<Order> reader,
                           ItemProcessor<Order, OrderRow> processor,
                           ItemWriter<OrderRow> writer) {
        return new StepBuilder("exportStep", jobRepository)
            .<Order, OrderRow>chunk(500, txManager) // chunk size is the commit interval — and a TX boundary
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .faultTolerant()
            .skip(FlatFileParseException.class)
            .skipLimit(50)
            .build();
    }
}

chunk(500, txManager) 的意思是：读取 500 个条目，逐个处理，再将这 500 个条目组成的列表交给写入器， 提交一个事务，然后重复。Chunk 是重启和回滚的单位。

幂等性与可重启性——最容易忽略的首要运维问题

JobInstance 由其标识性 JobParameters 确定。使用相同的标识性参数启动同一个作业两次，会得到：

JobInstanceAlreadyCompleteException: A job instance already exists and is complete

这是设计如此——Batch 会拒绝重新运行已完成的工作。处理方式有两种：

// Option A — RunIdIncrementer on the job (above) + JobLauncherApplicationRunner bumps run.id each launch.
// Option B — add a unique identifying parameter yourself when launching:
JobParameters params = new JobParametersBuilder()
    .addString("status", "COMPLETED")              // identifying — part of the instance key
    .addLong("run.id", System.currentTimeMillis()) // identifying & unique — makes each run a new instance
    .toJobParameters();

如果某个参数只是元数据，不应改变实例标识（例如要记录但不用于标识的请求 ID），可以在参数上将标识标记设为 false：

.addString("requestId", requestId, false) // non-identifying — excluded from the instance key

相反，使用相同参数重新启动失败的作业时，作业会恢复执行——跳过已完成的步骤，并从最后一个已提交的 Chunk 继续执行失败的步骤。这正是元数据表的用途。如果你希望失败后恢复执行，就不要总是传入唯一参数来破坏这一机制。

ItemReader——排序键与线程安全

@Bean
@StepScope // required: late-binds jobParameters at step execution, not context startup
public JpaPagingItemReader<Order> orderReader(
        EntityManagerFactory emf,
        @Value("#{jobParameters['status']}") String status) {
    return new JpaPagingItemReaderBuilder<Order>()
        .name("orderReader")
        .entityManagerFactory(emf)
        .queryString("SELECT o FROM Order o WHERE o.status = :status ORDER BY o.id") // ORDER BY is MANDATORY
        .parameterValues(Map.of("status", OrderStatus.valueOf(status)))
        .pageSize(500) // keep pageSize == chunk size
        .build();
}

• 分页读取器必须对唯一列使用确定性的 ORDER BY。 否则，数据库会在不同分页中以任意顺序返回行 → 行会被跳过或重复处理。这是无声的数据损坏，而不是错误。
• JdbcCursorItemReader 不是线程安全的。 JdbcPagingItemReader / JpaPagingItemReader 可用于多线程步骤。对于多线程步骤中的非线程安全读取器，请用 SynchronizedItemStreamReader 包装。
• 不要在同一个作业中修改用于分页的列。 如果写入器在读取器按 WHERE status = 'PENDING' ORDER BY id 分页时将 status 从 PENDING 改为 DONE，结果集会随之变化，导致漏页。读取稳定快照、按不可变的 id 分页，或使用游标读取器。

ItemProcessor——返回 null 会过滤条目

@Component
public class OrderProcessor implements ItemProcessor<Order, OrderRow> {
    @Override
    public OrderRow process(Order order) {
        if (order.getTotal().isZero()) {
            return null; // ⚠️ null = FILTER this item; it is NOT written and NOT an error
        }
        return OrderRow.from(order);
    }
}

返回 null 会静默地将条目从 Chunk 中移除。这是一个实用特性（过滤），但如果你意外返回了 null，还以为条目会原样传递，那就会变成一个陷阱。

ItemWriter——Spring Batch 5 的签名变更

在 5.x 中，写入器接收的是 Chunk<? extends T>，而不是 List<? extends T>：

@Override
public void write(Chunk<? extends OrderRow> chunk) { // was List<? extends T> in 4.x
    repository.saveAll(chunk.getItems());
}

对于 SQL 写入，优先使用批处理 JDBC 写入器，而不是逐行保存——它会使用一次 addBatch()：

@Bean
public JdbcBatchItemWriter<OrderRow> orderWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<OrderRow>()
        .dataSource(dataSource)
        .sql("INSERT INTO order_export (id, total) VALUES (:id, :total)")
        .beanMapped()
        .build();
}

写入器在Chunk 事务内运行。切勿从写入器发送电子邮件、发布到 Kafka 或调用 Webhook——如果 Chunk 回滚，消息也已经发出。应将副作用绑定到作业完成事件（参见 [[transactional-patterns]] 和下方的监听器）。

启动作业

默认情况下，Boot 会在启动时运行每个 Job Bean。对于计划任务或按需执行的作业，请关闭此功能并显式启动：

spring:
  batch:
    job:
      enabled: false           # don't run jobs on app startup; we trigger them ourselves
    jdbc:
      initialize-schema: never # manage BATCH_* tables with Flyway in prod (see below)

@Component
@RequiredArgsConstructor
public class OrderExportScheduler {

    private final JobLauncher jobLauncher;
    private final Job orderExportJob;

    @Scheduled(cron = "0 0 2 * * *")
    public void runNightly() throws JobExecutionException {
        JobParameters params = new JobParametersBuilder()
            .addString("status", "COMPLETED")
            .addLong("run.id", System.currentTimeMillis())
            .toJobParameters();
        jobLauncher.run(orderExportJob, params);
    }
}

默认的 JobLauncher 是同步的（SyncTaskExecutor）——jobLauncher.run(...) 会阻塞 @Scheduled 线程，直到整个作业完成。若要实现发起后即返回，请使用异步 TaskExecutor 配置启动器；或者仅在能接受阻塞时，才从请求线程触发作业。

生产环境中的元数据架构

Spring Batch 需要 BATCH_JOB_INSTANCE、BATCH_JOB_EXECUTION、BATCH_STEP_EXECUTION 等表。 initialize-schema: always 适用于开发环境或嵌入式数据库，但不要让 Batch 在启动时对生产数据库执行 DDL。设置 initialize-schema: never，并将架构作为带版本号的 [[flyway-migrations]] 迁移一起发布（规范 DDL 位于 spring-batch-core 中的 org/springframework/batch/core/schema-*.sql）。

监听器——必须在作业结束后运行，而不是每个块都运行的工作

@Bean
public Job orderExportJob(JobRepository jobRepository, Step exportStep) {
    return new JobBuilder("orderExportJob", jobRepository)
        .incrementer(new RunIdIncrementer())
        .listener(new JobExecutionListener() {
            @Override public void afterJob(JobExecution exec) {
                if (exec.getStatus() == BatchStatus.COMPLETED) {
                    notifier.notifyExportReady(exec.getJobParameters()); // safe: all chunks committed
                }
            }
        })
        .start(exportStep)
        .build();
}

不需要 Batch 时就别用它

对于大型、可重启、可审计的批量作业，Spring Batch 的复杂性（元数据表、重启、分块）才有意义。对于快速的一次性异步任务，@Async 或 @Scheduled 循环更轻量。对于需要持久化并支持重试的后台作业，作业队列更合适。根据规模选择合适的工具。

注意事项

• Agent 添加了 @EnableBatchProcessing——在 Boot 3 中它会禁用自动配置；移除它，只需注入 JobRepository
• Agent 使用了 JobBuilderFactory / StepBuilderFactory——它们已在 Batch 5 中移除；使用 new JobBuilder(name, repo) / new StepBuilder(name, repo)
• Agent 调用 .chunk(500) 时没有事务管理器——Batch 5 要求使用 .chunk(500, txManager)
• Agent 编写了 write(List<? extends T> items)——Batch 5 的签名是 write(Chunk<? extends T> chunk)
• Agent 使用相同参数重新运行作业并遇到 JobInstanceAlreadyCompleteException——添加 RunIdIncrementer 或唯一的标识参数
• Agent 为应该在失败后恢复的作业每次运行都添加唯一参数——这会破坏可重启性；只有需要创建新实例时才添加
• Agent 编写的分页读取器查询没有 ORDER BY（或排序列不唯一）——分页会悄无声息地跳过或重复行；按唯一列排序
• Agent 在多线程步骤中使用 JdbcCursorItemReader——它不是线程安全的；使用分页读取器或 SynchronizedItemStreamReader
• Agent 按作业中会被写入器修改的列分页——结果集会发生偏移；按不可变 ID 分页
• Agent 从期望直通的处理器返回 null——null 会过滤掉（丢弃）该项
• Agent 从 ItemWriter 发送电子邮件／发布事件——这些操作会在块事务中运行；改为在 afterJob 监听器中执行
• Agent 忘记为读取 jobParameters 的读取器添加 @StepScope——@Value("#{jobParameters[...]}") 只会在步骤作用域中绑定
• Agent 让作业在 Web 应用启动时运行——设置 spring.batch.job.enabled=false 并显式启动作业
• Agent 让 initialize-schema: always 对生产数据库执行 DDL——使用 never，并通过 Flyway 迁移创建 BATCH_* 表