Spring Batch大數(shù)據(jù)量處理之從入門到精通實踐

更新時間：2026年04月18日 14:33:29 作者：程序員鴨梨

文章介紹了SpringBatch的基本架構和核心配置,接著通過CSV導入數(shù)據(jù)庫和數(shù)據(jù)庫導出到文件兩個簡單任務示例進行講解,最后介紹了如何處理多步驟復雜任務

一、Spring Batch 基礎架構

1.1 核心配置

@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 數(shù)據(jù)庫表結構

-- Spring Batch 元數(shù)據(jù)表
-- BATCH_JOB_INSTANCE: 作業(yè)實例
-- BATCH_JOB_EXECUTION: 作業(yè)執(zhí)行
-- BATCH_JOB_EXECUTION_PARAMS: 作業(yè)參數(shù)
-- BATCH_STEP_EXECUTION: 步驟執(zhí)行
-- BATCH_JOB_EXECUTION_CONTEXT: 作業(yè)上下文
-- BATCH_STEP_EXECUTION_CONTEXT: 步驟上下文

-- 創(chuàng)建自定義監(jiān)控表
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 導入數(shù)據(jù)庫

@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)
            .<ProductInput, Product>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<ProductInput> csvItemReader() {
        return new FlatFileItemReaderBuilder<ProductInput>()
            .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<ProductInput, Product> 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<Product> databaseItemWriter() {
        return new JdbcBatchItemWriterBuilder<Product>()
            .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 數(shù)據(jù)庫導出到文件

@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)
            .<Order, OrderOutput>chunk(500, transactionManager)
            .reader(orderItemReader())
            .processor(orderItemProcessor())
            .writer(orderItemWriter())
            .build();
    }

    @Bean
    public JdbcPagingItemReader<Order> orderItemReader() {
        return new JdbcPagingItemReaderBuilder<Order>()
            .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<OrderOutput> orderItemWriter() {
        return new FlatFileItemWriterBuilder<OrderOutput>()
            .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 多步驟任務

@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)
            .<Order, ValidatedOrder>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)
            .<OrderItem, InventoryUpdate>chunk(200, transactionManager)
            .reader(orderItemReader())
            .processor(inventoryProcessor())
            .writer(inventoryWriter())
            .build();
    }

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

    @Bean
    public Flow flow1() {
        return new FlowBuilder<Simple