SpringBatch读取mysql数据

1.在本地数据库创建user表

建表语句：

create table `user`(
    `id` bigint not null auto_increment comment '主键',
    `name` varchar(32) default null comment '用户名',
    `age` int default null comment '年龄',
    primary key(id)
)engine=innodb  default  char set =utf8mb4;

新增数据：

insert into `user` (`name`,`age`) values
('dafei',15),('erfei',16),('sanfei',17),('sifei',18),('wufei',19);

示例：

需求：将SpringBatch库中user表的数据通过Springbatch读取到控制台中

SpringBatch提供了两种方法帮助我们从数据库中读取数据。

1.基于游标的方式

定义：一条一条的读数据，读完一条再读下一条。

流程：读数据=>封装=>读入

封装条件：如果我们使用JDBC读取数据，游标会将那条数据封装到ResultSet中，如果想从ResultSet中将数据读取出来，我们需要借助SpringBatch的RowMapper类实现表数据与实体对象的映射。

具体实现：

创建user对象

package com.pjk.springBatch.demo2;

import lombok.Data;

@Data
public class User {
    private Long id;
    private String name;
    private int age;
}

创建映射对象

package com.pjk.springBatch.demo2;

import org.springframework.jdbc.core.RowMapper;

import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * 创建映射对象
 */
public class UserRowMapper implements RowMapper<User> {
    @Override
    public User mapRow(ResultSet rs, int rowNum) throws SQLException {

        User user = new User();
        user.setId(rs.getLong("id"));
        user.setName(rs.getString("name"));
        user.setAge(rs.getInt("age"));
        return user;
    }
}

具体代码实现：

package com.pjk.springBatch.demo2;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.*;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

import javax.sql.DataSource;
import java.util.List;

@EnableBatchProcessing//3.开启springbatch
@SpringBootApplication//2.开启springboot
public class SpringBatchMysqlDemo {//1.创建类
    public static void main(String[] args) {
        SpringApplication.run(SpringBatchMysqlDemo.class, args);//4.创建启动主方法
    }
    @Autowired
    private JobBuilderFactory jobBuilderFactory;//5.创建job工厂

    @Autowired
    private StepBuilderFactory stepBuilderFactory;//6.创建step工厂

    @Autowired//18.创建数据源对象
    private DataSource dataSource;

    @Bean//21.创建映射对象
    public UserRowMapper userRowMapper() {
        return new UserRowMapper();
    }


    @Bean//23.创建写对象
    public ItemWriter<User> itemWrite() {
        return new ItemWriter<User>() {
            @Override
            public void write(List<? extends User> list) throws Exception {
                list.forEach(System.out::println);//24.写的具体逻辑
            }
        };
    }


    @Bean//15.创建读取数据对象
    public JdbcCursorItemReader<User> itemReader() {
        return new JdbcCursorItemReaderBuilder<User>()
                //16.起个名字
                .name("userItemReader")
                //17.连接数据库
                .dataSource(dataSource)
                //19.执行sql语句
                .sql("select * from user")//以游标的方式一条一条的从user中读取数据
                //20.指定映射对象
                .rowMapper(userRowMapper())
                .build();


    }


    @Bean//10.创建一个step
    public Step step() {
        return stepBuilderFactory.get("stepOne")//11.给step起名字
                .allowStartIfComplete(true)//12.允许step运行完之后 可以再次运行
                .<User,User>chunk(1)//13.开启块处理模式
                .reader(itemReader())//14.块处理中第一步读取数据
                .writer(itemWrite())//22.块处理中第二步写数据
                .build();
    }
    
    @Bean//7.创建job
    public Job job() {
        return jobBuilderFactory.get("job")//8.给job起名字
                .start(step())//9.执行第一个任务
                .incrementer(new RunIdIncrementer())//25.允许job多次启动不手动修改job名
                .build();
    }
}

结果展示：