Spring Batch 数据处理：2025 实战指南

Spring Batch 数据处理2025 实战指南我是 Alex一个在 CSDN 写 Java 架构思考的暖男。看到新手博主写技术踩坑记录总会留言这个 debug 思路很 solid下次试试加个 circuit breaker 会更优雅。我的文章里从不说空话每个架构图都经过生产环境验证。对了别叫我大神喊我 Alex 就好。一、Spring Batch 核心概念Spring Batch 是 Spring 生态系统中用于批处理的框架它提供了强大的批处理功能支持大规模数据处理。1.1 核心概念Job批处理作业是批处理的顶层概念Step作业的步骤一个作业由一个或多个步骤组成ItemReader读取数据的组件ItemProcessor处理数据的组件ItemWriter写入数据的组件JobRepository存储作业执行状态的仓库JobLauncher启动作业的组件JobExecution作业执行实例StepExecution步骤执行实例1.2 Spring Batch 的优势可扩展性支持大规模数据处理可靠性支持事务管理和重启机制可监控性提供详细的执行状态和日志灵活性支持多种数据源和处理方式集成性与 Spring 生态系统无缝集成二、Spring Batch 配置2.1 基本配置Configuration EnableBatchProcessing public class BatchConfig { Autowired private JobBuilderFactory jobBuilderFactory; Autowired private StepBuilderFactory stepBuilderFactory; Bean public ItemReaderUser userItemReader() { // 从数据库读取数据 return new JdbcCursorItemReaderBuilderUser() .name(userItemReader) .dataSource(dataSource) .sql(SELECT id, name, email FROM users WHERE status ACTIVE) .rowMapper(new BeanPropertyRowMapper(User.class)) .build(); } Bean public ItemProcessorUser, UserDTO userItemProcessor() { return user - { UserDTO dto new UserDTO(); dto.setId(user.getId()); dto.setName(user.getName().toUpperCase()); dto.setEmail(user.getEmail().toLowerCase()); return dto; }; } Bean public ItemWriterUserDTO userItemWriter() { // 写入到文件 return items - { for (UserDTO item : items) { System.out.println(Processing user: item.getName()); // 写入到文件或其他目标 } }; } Bean public Step processUserStep() { return stepBuilderFactory.get(processUserStep) .User, UserDTOchunk(10) .reader(userItemReader()) .processor(userItemProcessor()) .writer(userItemWriter()) .build(); } Bean public Job processUserJob() { return jobBuilderFactory.get(processUserJob) .incrementer(new RunIdIncrementer()) .flow(processUserStep()) .end() .build(); } }2.2 数据源配置Configuration public class DataSourceConfig { Bean public DataSource dataSource() { HikariConfig config new HikariConfig(); config.setJdbcUrl(jdbc:mysql://localhost:3306/batch_db); config.setUsername(root); config.setPassword(password); config.setMaximumPoolSize(10); return new HikariDataSource(config); } Bean public JdbcTemplate jdbcTemplate(DataSource dataSource) { return new JdbcTemplate(dataSource); } }2.3 作业仓库配置Configuration public class JobRepositoryConfig { Bean public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception { JobRepositoryFactoryBean factory new JobRepositoryFactoryBean(); factory.setDataSource(dataSource); factory.setTransactionManager(transactionManager); factory.setIsolationLevelForCreate(ISOLATION_SERIALIZABLE); factory.setTablePrefix(BATCH_); factory.setMaxVarCharLength(1000); return factory.getObject(); } Bean public PlatformTransactionManager transactionManager(DataSource dataSource) { return new DataSourceTransactionManager(dataSource); } Bean public JobLauncher jobLauncher(JobRepository jobRepository) throws Exception { SimpleJobLauncher launcher new SimpleJobLauncher(); launcher.setJobRepository(jobRepository); launcher.setTaskExecutor(new SimpleAsyncTaskExecutor()); return launcher; } }三、ItemReader 实现3.1 数据库读取Bean public ItemReaderCustomer customerItemReader(DataSource dataSource) { return new JdbcPagingItemReaderBuilderCustomer() .name(customerItemReader) .dataSource(dataSource) .selectClause(SELECT id, first_name, last_name, email, phone) .fromClause(FROM customers) .whereClause(WHERE last_updated :lastUpdated) .parameterValues(Collections.singletonMap(lastUpdated, LocalDateTime.now().minusDays(1))) .sortKeys(Collections.singletonMap(id, Order.ASCENDING)) .rowMapper(new BeanPropertyRowMapper(Customer.class)) .pageSize(100) .build(); }3.2 文件读取Bean public ItemReaderProduct productItemReader() { return new FlatFileItemReaderBuilderProduct() .name(productItemReader) .resource(new ClassPathResource(products.csv)) .delimited() .names(id, name, price, quantity) .fieldSetMapper(fieldSet - { Product product new Product(); product.setId(fieldSet.readLong(id)); product.setName(fieldSet.readString(name)); product.setPrice(fieldSet.readBigDecimal(price)); product.setQuantity(fieldSet.readInt(quantity)); return product; }) .build(); }3.3 自定义读取器public class CustomItemReader implements ItemReaderString { private final ListString items; private int index 0; public CustomItemReader(ListString items) { this.items items; } Override public String read() { if (index items.size()) { return items.get(index); } return null; } } Bean public ItemReaderString customItemReader() { ListString items Arrays.asList(item1, item2, item3, item4, item5); return new CustomItemReader(items); }四、ItemProcessor 实现4.1 基本处理器public class ProductProcessor implements ItemProcessorProduct, ProductDTO { Override public ProductDTO process(Product item) { ProductDTO dto new ProductDTO(); dto.setId(item.getId()); dto.setName(item.getName()); dto.setPrice(item.getPrice()); dto.setQuantity(item.getQuantity()); dto.setTotalValue(item.getPrice().multiply(BigDecimal.valueOf(item.getQuantity()))); return dto; } } Bean public ItemProcessorProduct, ProductDTO productProcessor() { return new ProductProcessor(); }4.2 条件处理public class OrderProcessor implements ItemProcessorOrder, Order { Override public Order process(Order item) { if (item.getStatus().equals(OrderStatus.PENDING)) { item.setStatus(OrderStatus.PROCESSED); item.setProcessedAt(LocalDateTime.now()); return item; } return null; // 跳过非待处理订单 } } Bean public ItemProcessorOrder, Order orderProcessor() { return new OrderProcessor(); }4.3 复合处理器public class CompositeItemProcessorT, R implements ItemProcessorT, R { private final ListItemProcessor processors; public CompositeItemProcessor(ListItemProcessor processors) { this.processors processors; } Override public R process(T item) { Object result item; for (ItemProcessor processor : processors) { result processor.process(result); if (result null) { return null; } } return (R) result; } } Bean public ItemProcessorCustomer, CustomerDTO customerProcessor() { ListItemProcessor processors new ArrayList(); processors.add(new ValidationProcessor()); processors.add(new TransformationProcessor()); processors.add(new EnrichmentProcessor()); return new CompositeItemProcessor(processors); }五、ItemWriter 实现5.1 数据库写入Bean public ItemWriterCustomerDTO customerItemWriter(DataSource dataSource) { return new JdbcBatchItemWriterBuilderCustomerDTO() .dataSource(dataSource) .sql(INSERT INTO customer_processed (id, first_name, last_name, email, processed_at) VALUES (:id, :firstName, :lastName, :email, :processedAt)) .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider()) .build(); }5.2 文件写入Bean public ItemWriterProductDTO productItemWriter() { return new FlatFileItemWriterBuilderProductDTO() .name(productItemWriter) .resource(new FileSystemResource(output/products-processed.csv)) .delimited() .names(id, name, price, quantity, totalValue) .headerCallback(writer - writer.write(ID,Name,Price,Quantity,Total Value)) .build(); }5.3 自定义写入器public class CustomItemWriter implements ItemWriterUser { private final Logger logger LoggerFactory.getLogger(CustomItemWriter.class); Override public void write(List? extends User items) { for (User item : items) { logger.info(Writing user: {}, item.getName()); // 写入到外部系统或其他目标 } } } Bean public ItemWriterUser customItemWriter() { return new CustomItemWriter(); }六、作业执行与监控6.1 作业启动Service public class BatchService { Autowired private JobLauncher jobLauncher; Autowired private Job processUserJob; public void runProcessUserJob() throws Exception { JobParameters jobParameters new JobParametersBuilder() .addString(jobName, processUserJob) .addLong(time, System.currentTimeMillis()) .toJobParameters(); JobExecution execution jobLauncher.run(processUserJob, jobParameters); System.out.println(Job execution status: execution.getStatus()); } }6.2 作业监控RestController RequestMapping(/api/batch) public class BatchController { Autowired private JobExplorer jobExplorer; GetMapping(/jobs) public ListJobInstance getJobs() { return jobExplorer.getJobInstances(processUserJob, 0, 10); } GetMapping(/executions/{jobInstanceId}) public ListJobExecution getExecutions(PathVariable Long jobInstanceId) { JobInstance jobInstance jobExplorer.getJobInstance(jobInstanceId); return jobExplorer.getJobExecutions(jobInstance); } GetMapping(/steps/{jobExecutionId}) public ListStepExecution getSteps(PathVariable Long jobExecutionId) { JobExecution jobExecution jobExplorer.getJobExecution(jobExecutionId); return jobExecution.getStepExecutions(); } }6.3 作业调度Configuration EnableScheduling public class BatchScheduler { Autowired private JobLauncher jobLauncher; Autowired private Job processUserJob; Scheduled(cron 0 0 0 * * ?) // 每天凌晨执行 public void runDailyJob() throws Exception { JobParameters jobParameters new JobParametersBuilder() .addString(jobName, processUserJob) .addLong(time, System.currentTimeMillis()) .toJobParameters(); jobLauncher.run(processUserJob, jobParameters); } }七、Spring Batch 最佳实践7.1 性能优化合理设置 chunk 大小根据数据量和系统资源设置合适的 chunk 大小使用并行处理对于大规模数据处理使用并行步骤优化数据库操作使用批量操作减少数据库连接次数使用异步处理对于IO密集型操作使用异步处理7.2 错误处理跳过策略设置合理的跳过策略处理错误数据重试机制对于临时错误使用重试机制错误日志详细记录错误信息便于排查死信队列将无法处理的数据放入死信队列Bean public Step processOrderStep() { return stepBuilderFactory.get(processOrderStep) .Order, Orderchunk(10) .reader(orderItemReader()) .processor(orderItemProcessor()) .writer(orderItemWriter()) .faultTolerant() .skipLimit(10) .skip(OrderProcessingException.class) .retryLimit(3) .retry(ConnectionException.class) .build(); }7.3 事务管理合理设置事务边界根据业务需求设置合适的事务边界使用局部事务对于不需要全局事务的步骤使用局部事务事务隔离级别根据业务需求设置合适的事务隔离级别7.4 监控与告警作业执行监控监控作业执行状态和性能错误告警对作业执行错误进行告警性能指标收集作业执行的性能指标八、生产环境案例分析8.1 案例一电商平台数据同步某电商平台使用 Spring Batch 实现了从线下系统到线上系统的数据同步。主要功能包括从线下数据库读取商品信息处理和转换数据格式写入到线上数据库生成同步报告通过 Spring Batch该平台实现了每天同步超过 100 万条商品数据同步时间从原来的 4 小时减少到 30 分钟数据准确率达到 99.99%。8.2 案例二金融系统批处理某银行使用 Spring Batch 实现了每日批处理作业包括账户余额计算交易对账报表生成风险评估通过 Spring Batch该银行实现了每天处理超过 1000 万笔交易批处理时间从原来的 6 小时减少到 1.5 小时系统稳定性显著提高。九、常见误区与解决方案9.1 内存溢出问题处理大量数据时出现内存溢出解决方案合理设置 chunk 大小使用分页读取避免一次性加载所有数据9.2 事务管理不当问题事务范围过大导致锁定时间过长解决方案合理设置事务边界使用局部事务9.3 错误处理不完善问题错误处理机制不完善导致作业频繁失败解决方案设置合理的跳过策略和重试机制9.4 监控不足问题缺乏对作业执行状态的监控解决方案建立完善的监控体系及时发现和解决问题十、总结与展望Spring Batch 是一个强大的批处理框架它为企业级应用提供了可靠、高效的数据处理能力。通过合理配置和使用 Spring Batch可以显著提高数据处理效率减少人工干预提高系统可靠性。在云原生时代Spring Batch 也在不断演进。未来我们将看到 Spring Batch 与云原生技术的深度融合如与 Kubernetes 的集成以及对 Serverless 架构的支持为批处理作业提供更加灵活、高效的运行环境。记住批处理作业的设计应该根据业务需求和数据特点进行合理规划。这其实可以更优雅一点。别叫我大神叫我 Alex 就好。如果你在 Spring Batch 实践中遇到了问题欢迎在评论区留言我会尽力为你提供建设性的建议。