Spring Batch批处理框架操作指南

简介

Spring Batch是针对于处理海量数据的批处理框架，它通过创建批处理作业（Job）来完成数据的读取、处理和写入输出的操作。本指南为您提供Spring Batch的完整操作指南。

原理

• Job: 对整个批处理过程进行定义与配置。
• Step: 是Job的一部分，代表一个特定的处理阶段。
• ItemReader: 用于读取数据。
• ItemProcessor: 用于对读取的数据进行处理。
• ItemWriter: 用于将处理后的数据写入到某个数据源中。
• JobLauncher: 用于启动Job。

步骤

1. 引入Spring Batch的依赖

<dependency>
    <groupId>org.springframework.batch</groupId>
    <artifactId>spring-batch-core</artifactId>
    <version>4.2.4.RELEASE</version>
</dependency>

2. 定义Job

通过Job接口来定义批处理任务，并通过Step接口来定义具体的处理步骤。

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private JobCompletionNotificationListener listener;

    @Bean
    public Job importJob() {
        return jobBuilderFactory.get("importJob")
                .incrementer(new RunIdIncrementer())
                .listener(listener)
                .flow(step1())
                .end()
                .build();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                .<Person, Person>chunk(10)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .build();
    }

    @Bean
    public ItemReader<Person> reader() {
        // ... 在这里实现读取数据的逻辑
    }

    @Bean
    public ItemProcessor<Person, Person> processor() {
        // ... 在这里实现数据处理的逻辑
    }

    @Bean
    public ItemWriter<Person> writer() {
        // ... 在这里实现数据写入的逻辑
    }
}

3. 实现读取逻辑

通过实现ItemReader接口来读取数据。

@Component
public class PersonItemReader implements ItemReader<Person> {

    private List<Person> persons;

    private int currIndex = 0;

    @PostConstruct
    public void init() {
        persons = new ArrayList<Person>();
        // ... 将数据存储到persons中
    }

    @Override
    public Person read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
        if (currIndex < persons.size()) {
            return persons.get(currIndex++);
        } else {
            return null;
        }
    }
}

4. 实现数据处理逻辑

通过实现ItemProcessor接口来对读取的数据进行处理。

示例：将读取到的Person对象的名字全部转成大写。

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

    @Override
    public Person process(Person person) throws Exception {
        String upperName = person.getName().toUpperCase();
        Person transformedPerson = new Person(upperName, person.getAge());
        return transformedPerson;
    }
}

5. 实现数据写入逻辑

通过实现ItemWriter接口来将处理后的数据写入到某个数据源中。

示例：将处理后的Person对象输出到控制台。

@Component
public class PersonItemWriter implements ItemWriter<Person> {

    @Override
    public void write(List<? extends Person> items) throws Exception {
        for (Person item : items) {
            System.out.println(item);
        }
    }
}

6. 实现JobLauncher的启动

通过实现JobLauncher接口来启动Job。

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    // ...

    @Autowired
    private Job importJob;

    @Autowired
    private JobLauncher jobLauncher;

    @GetMapping("/startJob")
    public String startJob() throws Exception {
        JobParameters jobParameters = new JobParametersBuilder()
                .addString("JobID", String.valueOf(System.currentTimeMillis()))
                .toJobParameters();
        jobLauncher.run(importJob, jobParameters);
        return "Success";
    }
}

示例

示例1：读取数据存储到数据库

该示例演示了如何将批量读取的数据存储到数据库中。

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    // 定义数据源
    @Bean
    public DataSource dataSource() {
        BasicDataSource dataSource = new BasicDataSource();
        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
        dataSource.setUrl("jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai");
        dataSource.setUsername("root");
        dataSource.setPassword("root");
        return dataSource;
    }

    // ...

    @Bean
    public PersonItemWriter writer() {
        JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<>();
        writer.setDataSource(dataSource());
        writer.setSql("INSERT INTO person (name, age) VALUES (:name, :age)");
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<Person>());
        return writer;
    }
}

示例2：多线程读取数据

该示例演示了如何使用多线程的方式来读取数据。

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    // ...

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                .<Person, Person>chunk(10)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .taskExecutor(taskExecutor())
                .build();
    }

    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(30);
        executor.setKeepAliveSeconds(60);
        executor.initialize();
        return executor;
    }
}