读数据——从数据库中批量读取数据

下面以mysql为例，用spring batch批量从mysql数据库表users表中读取数据。

uers表目前数据如下：



下面用spring batch批量读取该表，首先配置数据源

#配置step执行多少次commit一次
spring.batch.chunk.size=4
#数据源配置
spring.datasource.url=jdbc:mysql://localhost:3306/lzj_database
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
spring.datasource.username=root
spring.datasource.password=lzjlzj

配置读取user表的job作业

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import javax.sql.DataSource;
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.item.ItemReader;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.support.MySqlPagingQueryProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.RowMapper;

@EnableBatchProcessing
@Configuration
public class BatchConfig {

@Autowired
private StepBuilderFactory stepBuilderFactory;

@Autowired
private JobBuilderFactory jobBuilderFactory;

@Autowired
private DataSource dataSource;

@Value("${spring.batch.chunk.size:3}")
private int chunkSize;

/*1、创建一个Job作业*/
@Bean
public Job databaseJob(){
return jobBuilderFactory.get("dataBaseJob")
.start(chunkStep())
.build();
}

//2、创建一个step*/
@Bean
public Step chunkStep(){
return stepBuilderFactory.get("chunkStep")
.chunk(chunkSize)                                //每chunkSize次提交一次
.reader(databaseItemReader())                    //读取数据库，并把库表中每列数据映射到工程中的User bean中
.writer(list -> list.forEach(System.out::println))
.allowStartIfComplete(true)
.build();
}

/*3、读数据库配置*/
@Bean
public ItemReader<User> databaseItemReader(){
JdbcPagingItemReader<User> reader = new JdbcPagingItemReader<>();
reader.setDataSource(dataSource);               //设置数据源
reader.setFetchSize(2);                         //FetchSize设置为2，表示每次从数据库中,2条数据
reader.setRowMapper(new UserRowMapper());       //把数据库表中每条数据映射到User对象中
MySqlPagingQueryProvider queryProvider = new MySqlPagingQueryProvider();
queryProvider.setSelectClause("*");            //设置查询的列
queryProvider.setFromClause("from users");     //设置查询的表
Map<String, Order> sortKeys = new HashMap<>(); //定义一个map，用于存放排序列
sortKeys.put("id", Order.ASCENDING);           //按id列升序排列
sortKeys.put("age", Order.DESCENDING);         //按age的降序排列
queryProvider.setSortKeys(sortKeys);           //设置排序列
reader.setQueryProvider(queryProvider);
return reader;
}

/*把数据库表中每条数据映射到User对象中*/
public class UserRowMapper implements RowMapper<User>{
@Override
/*ResultSet数据库一条结果集；i表示当前行*/
public User mapRow(ResultSet resultSet, int i) throws SQLException {
return new User(resultSet.getInt("id"), resultSet.getString("name"),
resultSet.getInt("age"));
}

}

}

运行该作业，输出内容如下：

……省略
User [id=1, name=Bob, age=15]
User [id=2, name=Tom, age=10]
User [id=3, name=lzj, age=25]
User [id=4, name=zhangsan, age=21]
User [id=5, name=zhangsan, age=21]
User [id=6, name=lisi, age=20]
User [id=7, name=lisi, age=20]
User [id=24, name=wanger, age=25]
User [id=25, name=wanger, age=25]
User [id=26, name=wanger, age=25]
User [id=27, name=wanger, age=25]
User [id=28, name=wanger, age=25]
User [id=29, name=wanger, age=25]
User [id=30, name=wanger, age=25]
User [id=31, name=wanger, age=25]
User [id=32, name=wanger, age=25]
User [id=33, name=wanger, age=25]

运行该作业后，会在数据库中生成Job的执行信息，生成表如下



查看batch_step_execution表



可以看出，共17条数据，commit了5次。因为chunkSize设置的是4，需要5次chunk才能执行完毕。