spring-batch

Spring Batch

完整参考：跳过/重试配置、分区、监听器、使用 JobLauncherTestUtils 进行测试、组合写入器和异步处理，请参阅 advanced.md。

深入了解：使用 mcp__documentation__fetch_docs，技术指定为：spring-boot，主题指定为：batch，以获取全面的文档。

快速入门

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Bean
    public Job importJob(JobRepository jobRepository, Step step1) {
        return new JobBuilder("importJob", jobRepository)
            .incrementer(new RunIdIncrementer())
            .start(step1)
            .build();
    }

    @Bean
    public Step step1(JobRepository jobRepository,
                      PlatformTransactionManager transactionManager,
                      ItemReader<InputData> reader,
                      ItemProcessor<InputData, OutputData> processor,
                      ItemWriter<OutputData> writer) {
        return new StepBuilder("step1", jobRepository)
            .<InputData, OutputData>chunk(100, transactionManager)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .build();
    }
}

---

核心组件

Job

@Bean
public Job complexJob(JobRepository jobRepository,
                      Step extractStep, Step transformStep, Step loadStep) {
    return new JobBuilder("etlJob", jobRepository)
        .incrementer(new RunIdIncrementer())
        .validator(jobParametersValidator())
        .listener(jobExecutionListener())
        .start(extractStep)
        .next(transformStep)
        .next(loadStep)
        .build();
}

// Job with decision
@Bean
public Job conditionalJob(JobRepository jobRepository,
                          Step step1, Step step2, Step errorStep,
                          JobExecutionDecider decider) {
    return new JobBuilder("conditionalJob", jobRepository)
        .start(step1)
        .next(decider)
        .on("COMPLETED").to(step2)
        .from(decider).on("FAILED").to(errorStep)
        .end()
        .build();
}

Step - 分块处理

@Bean
public Step chunkStep(JobRepository jobRepository,
                      PlatformTransactionManager txManager) {
    return new StepBuilder("chunkStep", jobRepository)
        .<Person, Person>chunk(100, txManager)
        .reader(reader())
        .processor(processor())
        .writer(writer())
        .faultTolerant()
        .skipLimit(10)
        .skip(ValidationException.class)
        .retryLimit(3)
        .retry(TransientException.class)
        .build();
}

// Tasklet Step (for simple operations)
@Bean
public Step taskletStep(JobRepository jobRepository,
                        PlatformTransactionManager txManager) {
    return new StepBuilder("taskletStep", jobRepository)
        .tasklet((contribution, chunkContext) -> {
            cleanupService.cleanup();
            return RepeatStatus.FINISHED;
        }, txManager)
        .build();
}

---

ItemReader

FlatFileItemReader

@Bean
public FlatFileItemReader<Person> csvReader() {
    return new FlatFileItemReaderBuilder<Person>()
        .name("personReader")
        .resource(new ClassPathResource("data/input.csv"))
        .delimited()
        .delimiter(",")
        .names("firstName", "lastName", "email", "age")
        .linesToSkip(1)  // Skip header
        .fieldSetMapper(new BeanWrapperFieldSetMapper<>() {{
            setTargetType(Person.class);
        }})
        .build();
}

JdbcPagingItemReader

@Bean
public JdbcPagingItemReader<Person> pagingReader(DataSource dataSource) {
    Map<String, Order> sortKeys = new HashMap<>();
    sortKeys.put("id", Order.ASCENDING);

    return new JdbcPagingItemReaderBuilder<Person>()
        .name("pagingReader")
        .dataSource(dataSource)
        .selectClause("SELECT id, first_name, last_name, email")
        .fromClause("FROM persons")
        .whereClause("WHERE status = :status")
        .parameterValues(Map.of("status", "ACTIVE"))
        .sortKeys(sortKeys)
        .rowMapper(new BeanPropertyRowMapper<>(Person.class))
        .pageSize(100)
        .build();
}

JpaPagingItemReader

@Bean
public JpaPagingItemReader<Person> jpaReader(EntityManagerFactory emf) {
    return new JpaPagingItemReaderBuilder<Person>()
        .name("jpaReader")
        .entityManagerFactory(emf)
        .queryString("SELECT p FROM Person p WHERE p.status = :status")
        .parameterValues(Map.of("status", Status.ACTIVE))
        .pageSize(100)
        .build();
}

---

ItemProcessor

@Component
public class PersonProcessor implements ItemProcessor<Person, Person> {

    @Override
    public Person process(Person person) throws Exception {
        // Return null to filter out item
        if (!isValid(person)) {
            return null;
        }

        // Transform
        person.setEmail(person.getEmail().toLowerCase());
        person.setFullName(person.getFirstName() + " " + person.getLastName());
        return person;
    }
}

// Composite processor
@Bean
public CompositeItemProcessor<Person, Person> compositeProcessor() {
    return new CompositeItemProcessorBuilder<Person, Person>()
        .delegates(List.of(
            validationProcessor(),
            transformationProcessor(),
            enrichmentProcessor()
        ))
        .build();
}

---

ItemWriter

JdbcBatchItemWriter

@Bean
public JdbcBatchItemWriter<Person> jdbcWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<Person>()
        .dataSource(dataSource)
        .sql("INSERT INTO persons (first_name, last_name, email) VALUES (:firstName, :lastName, :email)")
        .beanMapped()
        .build();
}

JpaItemWriter

@Bean
public JpaItemWriter<Person> jpaWriter(EntityManagerFactory emf) {
    JpaItemWriter<Person> writer = new JpaItemWriter<>();
    writer.setEntityManagerFactory(emf);
    writer.setUsePersist(true);  // false = merge
    return writer;
}

FlatFileItemWriter

@Bean
public FlatFileItemWriter<Person> csvWriter() {
    return new FlatFileItemWriterBuilder<Person>()
        .name("personWriter")
        .resource(new FileSystemResource("output/persons.csv"))
        .delimited()
        .delimiter(",")
        .names("firstName", "lastName", "email")
        .headerCallback(writer -> writer.write("First Name,Last Name,Email"))
        .build();
}

---

作业参数

@Bean
@StepScope
public FlatFileItemReader<Person> parameterizedReader(
        @Value("#{jobParameters['inputFile']}") String inputFile) {
    return new FlatFileItemReaderBuilder<Person>()
        .name("reader")
        .resource(new FileSystemResource(inputFile))
        .delimited()
        .names("firstName", "lastName", "email")
        .targetType(Person.class)
        .build();
}

// Running job with parameters
@Service
@RequiredArgsConstructor
public class JobLauncherService {

    private final JobLauncher jobLauncher;
    private final Job importJob;

    public void runJob(String inputFile, LocalDate date) throws Exception {
        JobParameters params = new JobParametersBuilder()
            .addString("inputFile", inputFile)
            .addLocalDate("date", date)
            .addLong("timestamp", System.currentTimeMillis())
            .toJobParameters();

        JobExecution execution = jobLauncher.run(importJob, params);
        log.info("Job status: {}", execution.getStatus());
    }
}

---

不适用此技能的情况

• 实时处理 - 使用流处理（Kafka Streams、Flink）
• 简单的定时任务 - 改用 spring-scheduling
• 微服务数据同步 - 考虑使用消息传递实现事件驱动
• 小型数据集 - 批处理的开销可能不值得

反模式

反模式	问题	解决方案
分块大小过大	内存问题、事务耗时长	调整分块大小（100-1000）
未配置跳过策略	单个错误就会导致作业停止	为预期错误配置跳过策略
有状态的 ItemProcessor	线程安全问题	让处理器保持无状态
忽略作业参数	无法重启失败的作业	包含标识参数
未进行监控	故障无提示	配置 JobExecutionListener
大数据量时使用单线程	处理速度慢	使用分区

快速故障排除

问题	诊断	修复方法
作业未启动	检查作业仓库	验证数据库架构
作业重启失败	检查作业参数	添加 RunIdIncrementer
分块处理缓慢	检查提交间隔	调整分块大小
内存问题	监控堆内存使用情况	减小分块大小、以流方式处理数据
跳过策略未生效	检查跳过策略	配置可跳过的异常

最佳实践

• ✅ 对大量数据使用分块处理
• ✅ 配置跳过/重试机制以实现容错
• ✅ 使用分区实现并行处理
• ✅ 实现监听器以进行监控
• ✅ 使用 JobLauncherTestUtils 进行测试
• ❌ 不要将分块大小设置得过大
• ❌ 不要悄无声息地忽略错误
• ❌ 不要忘记为重启设置作业参数

参考文档

• Spring Batch