Spring Batch批处理框架实战教程-源码库

Spring Batch批处理框架实战教程：从零开始构建数据批处理任务

作为一名长期在企业级应用开发中摸爬滚打的开发者，我深知数据批处理在业务系统中的重要性。今天我想和大家分享我在Spring Batch框架上的实战经验，这个框架彻底改变了我对批处理任务的认知——从繁琐的手工处理到优雅的自动化流程。

环境准备与项目搭建

首先我们需要创建一个Spring Boot项目并引入Spring Batch依赖。记得我第一次配置时，因为漏掉了数据库依赖导致任务无法持久化，白白浪费了一个下午排查问题。



    org.springframework.boot
    spring-boot-starter-batch


    com.h2database
    h2
    runtime

构建第一个批处理任务

让我们从一个简单的CSV文件数据导入任务开始。这个场景我在电商数据迁移项目中经常遇到，通过Reader、Processor、Writer的清晰分工，代码可维护性大大提升。


@Configuration
@EnableBatchProcessing
public class BatchConfig {
    
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    
    @Bean
    public FlatFileItemReader reader() {
        return new FlatFileItemReaderBuilder()
            .name("productItemReader")
            .resource(new ClassPathResource("products.csv"))
            .delimited()
            .names("id", "name", "price")
            .targetType(Product.class)
            .build();
    }
    
    @Bean
    public ItemProcessor processor() {
        return product -> {
            // 价格增加10%的处理逻辑
            product.setPrice(product.getPrice() * 1.1);
            return product;
        };
    }
    
    @Bean
    public JdbcBatchItemWriter writer(DataSource dataSource) {
        return new JdbcBatchItemWriterBuilder()
            .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
            .sql("INSERT INTO product (id, name, price) VALUES (:id, :name, :price)")
            .dataSource(dataSource)
            .build();
    }
}

配置任务步骤与作业

配置Step和Job时，我建议一定要设置合适的chunk大小。经过多次性能测试，我发现chunk大小在10-100之间通常能获得最佳性能平衡。


@Bean
public Step importProductStep(ItemReader reader,
                             ItemProcessor processor,
                             ItemWriter writer) {
    return stepBuilderFactory.get("importProductStep")
        .chunk(50)
        .reader(reader)
        .processor(processor)
        .writer(writer)
        .build();
}

@Bean
public Job importProductJob(JobCompletionNotificationListener listener, 
                           Step importProductStep) {
    return jobBuilderFactory.get("importProductJob")
        .incrementer(new RunIdIncrementer())
        .listener(listener)
        .flow(importProductStep)
        .end()
        .build();
}

实战中的坑与解决方案

在我第一次在生产环境使用Spring Batch时，遇到了任务重复执行的问题。后来发现是因为没有正确配置JobRepository的持久化。这里分享我的解决方案：


@Configuration
public class BatchPersistenceConfig {
    
    @Bean
    public JobRepository jobRepository(DataSource dataSource, 
                                     PlatformTransactionManager transactionManager) 
        throws Exception {
        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        factory.setDataSource(dataSource);
        factory.setTransactionManager(transactionManager);
        factory.setIsolationLevelForCreate("ISOLATION_DEFAULT");
        factory.setTablePrefix("BATCH_");
        return factory.getObject();
    }
}

监控与错误处理

批处理任务的监控至关重要。我习惯使用Spring Batch提供的JobExecutionListener来记录任务执行情况，这对于后续的问题排查非常有帮助。


@Component
public class JobCompletionNotificationListener extends JobExecutionListenerSupport {
    
    private static final Logger log = LoggerFactory.getLogger(
        JobCompletionNotificationListener.class);
    
    @Override
    public void afterJob(JobExecution jobExecution) {
        if(jobExecution.getStatus() == BatchStatus.COMPLETED) {
            log.info("!!! 批处理任务完成！");
        } else {
            log.error("!!! 批处理任务失败: {}", jobExecution.getStatus());
        }
    }
}

通过这个完整的实战示例，相信你已经对Spring Batch有了深入的理解。记住，批处理任务的成功不仅在于代码的正确性，更在于对异常情况的充分考虑和业务场景的合理设计。希望我的经验能帮助你在批处理开发中少走弯路！