SpringBatch大数据量处理之从入门到精通实践

一、Spring Batch 基础架构

1.1 核心配置

搭建一个高效的批处理系统，第一步就是把基础设施搭牢。下面这段配置展示了Spring Batch的核心组件——JobRepository、事务管理器、JobLauncher等——如何通过Spring配置有机地组合在一起。注意这里用 TaskExecutorJobLauncher 配合异步执行器，能让任务跑起来更灵活；而 JobExplorer 和 JobRegistry 的引入，则为后续的监控和动态管理埋下了伏笔。

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Autowired
    private JobRepository jobRepository;

    @Autowired
    private PlatformTransactionManager transactionManager;

    @Bean
    public JobLauncher jobLauncher() throws Exception {
        TaskExecutorJobLauncher jobLauncher = new TaskExecutorJobLauncher();
        jobLauncher.setJobRepository(jobRepository);
        jobLauncher.setTaskExecutor(new SimpleAsyncTaskExecutor());
        jobLauncher.afterPropertiesSet();
        return jobLauncher;
    }

    @Bean
    public JobExplorer jobExplorer(DataSource dataSource) throws Exception {
        JobExplorerFactoryBean factoryBean = new JobExplorerFactoryBean();
        factoryBean.setDataSource(dataSource);
        factoryBean.afterPropertiesSet();
        return factoryBean.getObject();
    }

    @Bean
    public JobRegistry jobRegistry() {
        return new MapJobRegistry();
    }

    @Bean
    public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor() {
        JobRegistryBeanPostProcessor postProcessor = new JobRegistryBeanPostProcessor();
        postProcessor.setJobRegistry(jobRegistry());
        return postProcessor;
    }
}

1.2 数据库表结构

Spring Batch 的元数据表是它的“大脑”，记录着每一次作业的实例、执行、参数、步骤上下文等关键信息。不过，光靠这几张表还不够，我们经常需要自定义监控表来追踪业务维度的执行情况。下面这个 batch_job_monitoring 表就是典型的扩展——把读、写、跳过次数和错误信息都单独记下来，方便后续做性能分析或异常排查。

-- Spring Batch 元数据表
-- BATCH_JOB_INSTANCE: 作业实例
-- BATCH_JOB_EXECUTION: 作业执行
-- BATCH_JOB_EXECUTION_PARAMS: 作业参数
-- BATCH_STEP_EXECUTION: 步骤执行
-- BATCH_JOB_EXECUTION_CONTEXT: 作业上下文
-- BATCH_STEP_EXECUTION_CONTEXT: 步骤上下文

-- 创建自定义监控表
CREATE TABLE batch_job_monitoring (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    job_name VARCHAR(100) NOT NULL,
    job_instance_id BIGINT,
    job_execution_id BIGINT,
    start_time TIMESTAMP,
    end_time TIMESTAMP,
    status VARCHAR(20),
    read_count BIGINT DEFAULT 0,
    write_count BIGINT DEFAULT 0,
    skip_count BIGINT DEFAULT 0,
    error_message TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

二、简单任务示例

2.1 CSV 导入数据库

最常见的批处理场景之一，就是把 CSV 文件里的数据批量写入数据库。这里用 FlatFileItemReader 读取文件，按分块每 1000 条处理一次，配上 faultTolerant 做容错——允许跳过 10 条验证异常，遇到临时数据访问异常还可以重试 3 次。注意，写入时用了 ON DUPLICATE KEY UPDATE，这样即使数据重复，也能优雅地更新而非报错中断。

@Configuration
public class CsvToDatabaseJobConfig {

    @Autowired
    private JobRepository jobRepository;

    @Autowired
    private PlatformTransactionManager transactionManager;

    @Bean
    public Job csvToDatabaseJob() {
        return new JobBuilder("csvToDatabaseJob", jobRepository)
            .start(csvToDatabaseStep())
            .listener(jobExecutionListener())
            .build();
    }

    @Bean
    public Step csvToDatabaseStep() {
        return new StepBuilder("csvToDatabaseStep", jobRepository)
            .chunk(1000, transactionManager)
            .reader(csvItemReader())
            .processor(productItemProcessor())
            .writer(databaseItemWriter())
            .faultTolerant()
            .skipLimit(10)
            .skip(ValidationException.class)
            .retryLimit(3)
            .retry(TransientDataAccessException.class)
            .listener(stepExecutionListener())
            .build();
    }

    @Bean
    public FlatFileItemReader csvItemReader() {
        return new FlatFileItemReaderBuilder()
            .name("csvItemReader")
            .resource(new FileSystemResource("input/products.csv"))
            .delimited()
            .names("id", "name", "description", "price", "category")
            .fieldSetMapper(new BeanWrapperFieldSetMapper<>() {{
                setTargetType(ProductInput.class);
            }})
            .linesToSkip(1) // 跳过表头
            .build();
    }

    @Bean
    public ItemProcessor productItemProcessor() {
        return input -> {
            Product product = new Product();
            product.setId(input.getId());
            product.setName(input.getName().trim());
            product.setDescription(input.getDescription());
            product.setPrice(new BigDecimal(input.getPrice()));
            product.setCategory(input.getCategory());
            product.setCreatedAt(LocalDateTime.now());
            return product;
        };
    }

    @Bean
    public JdbcBatchItemWriter databaseItemWriter() {
        return new JdbcBatchItemWriterBuilder()
            .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
            .sql("INSERT INTO products (id, name, description, price, category, created_at) " +
                 "VALUES (:id, :name, :description, :price, :category, :createdAt) " +
                 "ON DUPLICATE KEY UPDATE " +
                 "name = VALUES(name), description = VALUES(description), " +
                 "price = VALUES(price), category = VALUES(category)")
            .dataSource(dataSource)
            .build();
    }
}

2.2 数据库导出到文件

反过来，把数据库里的订单数据导出到 CSV 文件也很常见。这里用了 JdbcPagingItemReader 分页读取，通过 PagingQueryProvider 自定义查询条件和排序。注意，分页查询时排序键是必须的，否则数据顺序无法保证。写入部分用 FlatFileItemWriter 并添加了表头和页脚，方便后续处理工具直接读取。

@Configuration
public class DatabaseToFileJobConfig {

    @Bean
    public Job exportOrdersJob() {
        return new JobBuilder("exportOrdersJob", jobRepository)
            .start(exportOrdersStep())
            .build();
    }

    @Bean
    public Step exportOrdersStep() {
        return new StepBuilder("exportOrdersStep", jobRepository)
            .chunk(500, transactionManager)
            .reader(orderItemReader())
            .processor(orderItemProcessor())
            .writer(orderItemWriter())
            .build();
    }

    @Bean
    public JdbcPagingItemReader orderItemReader() {
        return new JdbcPagingItemReaderBuilder()
            .name("orderItemReader")
            .dataSource(dataSource)
            .queryProvider(new PagingQueryProvider() {
                @Override
                public void init(DataSource dataSource) {}
                
                @Override
                public String getSortKey() {
                    return "id";
                }
                
                @Override
                public String getSelectClause() {
                    return "SELECT id, user_id, total_amount, status, created_at";
                }
                
                @Override
                public String getFromClause() {
                    return "FROM orders";
                }
                
                @Override
                public String getWhereClause() {
                    return "WHERE created_at >= :startDate AND created_at <= :endDate";
                }
            })
            .parameterValues(Map.of("startDate", startDate, "endDate", endDate))
            .pageSize(1000)
            .rowMapper(new OrderRowMapper())
            .build();
    }

    @Bean
    public FlatFileItemWriter orderItemWriter() {
        return new FlatFileItemWriterBuilder()
            .name("orderItemWriter")
            .resource(new FileSystemResource("output/orders.csv"))
            .delimited()
            .delimiter(",")
            .names("orderId", "userId", "amount", "status", "createdDate")
            .headerCallback(writer -> writer.write("OrderID,UserID,Amount,Status,CreatedDate"))
            .footerCallback(writer -> writer.write("Total records exported"))
            .build();
    }
}

三、复杂任务处理

3.1 多步骤任务

实际业务中的批处理很少是单一步骤的，往往需要串行、并行、条件分支等组合。下面这个订单处理流程就是一个典型的多步骤编排：先校验订单，再处理支付，接着更新库存，然后发送通知——如果某一步失败，就跳到错误处理步骤；通知成功后，最终执行清理步骤。注意这里用 splitFlow 实现了并行分叉，把三个子流程（flow1、flow2、flow3）并发执行，能显著提升吞吐量。

@Configuration
public class ComplexBatchJobConfig {

    @Bean
    public Job orderProcessingJob() {
        return new JobBuilder("orderProcessingJob", jobRepository)
            .start(validateOrderStep())
            .next(processPaymentStep())
            .next(updateInventoryStep())
            .next(sendNotificationStep())
            .on("FAILED").to(errorHandlingStep())
            .from(sendNotificationStep()).on("*").to(cleanupStep())
            .end()
            .build();
    }

    @Bean
    public Step validateOrderStep() {
        return new StepBuilder("validateOrderStep", jobRepository)
            .chunk(100, transactionManager)
            .reader(pendingOrderReader())
            .processor(orderValidator())
            .writer(validatedOrderWriter())
            .build();
    }

    @Bean
    public Step processPaymentStep() {
        return new StepBuilder("processPaymentStep", jobRepository)
            .tasklet((contribution, chunkContext) -> {
                // 处理支付逻辑
                JobParameters params = chunkContext.getStepContext().getJobParameters();
                String batchId = params.getString("batchId");
                
                paymentService.processBatchPayments(batchId);
                
                return RepeatStatus.FINISHED;
            }, transactionManager)
            .build();
    }

    @Bean
    public Step updateInventoryStep() {
        return new StepBuilder("updateInventoryStep", jobRepository)
            .chunk(200, transactionManager)
            .reader(orderItemReader())
            .processor(inventoryProcessor())
            .writer(inventoryWriter())
            .build();
    }

    @Bean
    public Flow splitFlow() {
        return new FlowBuilder("splitFlow")
            .split(taskExecutor())
            .add(flow1(), flow2(), flow3())
            .build();
    }

    @Bean
    public Flow flow1() {
        return new FlowBuilder