技术摘要：springboot线程池、mybatis

〇、场景

目标：短时间内实现百万级数据落库

txt文件格式，每行为一条数据，共200w条，如下：

|identity|name|age|address|desc|

|123123|张三|20|xxxxxxxxxxx|xxxxxx|

|321321|李四|22|xxxxxxxx||

|112233|王五|21|xxxxxxxxxxxx|xxxxxxxx|

......

一、流程

1. 读取文件

目标：将txt文件读取到系统中

通过IO流的方式读取文件，设置路径和编码格式

数据列数(column)：字段数，分隔符号(delimiter)，字符截取的时候会使用

//数据列数
int column = 5;
//分割符号
String delimiter = "|";
//文件绝对路径
String path = "C:\\Users\\Admin\\Desktop\\test.txt";
//编码格式
String encoding = "UTF-8";//IO流读取文件
File file = new File(path);
InputStreamReader in = new InputStreamReader(new FileInputStream(file),encoding);
BufferedReader read = new BufferedReader(in);

2. 字符截取将数据存入集合

目标：将数据流存储到到实体列表中

①得到分隔符(delimiter)的索引数组：将单条数据分割成单个字符存到数组(charArr)，遍历charArr将delimiter的索引存到数组(indexArr)

②根据indexArr就可以截取分隔符之间的数据，得到数据数组(strArr)

③遍历strArr将数据存在实体(person)中，然后再add到集合(personList)中

注：为什么不直接用lineTxt.split("|")，因为数据存在空值“||”分隔符间没有数据会导致不正确的截取

String lineTxt = null;
List<Person> personList = new ArrayList<>();
while((lineTxt = read.readLine()) != null){//分割为字符数组String[] charArr = lineTxt.split("");//存储“|”在数组中的的indexint[] indexArr = new int[column+1];int j = 0;for (int i = 0; i < charArr.length; i++) {if(delimiter.equals(charArr[i])){indexArr[j] = i;j++;}}//存储“|”与“|”之间的数据，如果没有则为nullString[] strArr = new String[column];int k = 0;for (int i = 0; i < indexArr.length; i++) {if(i < indexArr.length-1){//截取两个相邻index之间的字符并去掉两端的空格，如果为”||“strArr[k] = lineTxt.substring(indexArr[i]+1,indexArr[i+1]).trim();k++;}}//存入实体类Person person = new Person();for (int i = 0; i < strArr.length; i++) {person.setId(UUID.randomUUID().toString());person.setIdentity(strArr[0]);person.setName(strArr[1]);person.setAge(strArr[2]);person.setAddress(strArr[3]);person.setDesc(strArr[4]);}//添加到集合personList.add(person);
}

3. 数组截取并落库

目标：实现分批落库

得到长度为200w的集合后，不能直接批量插入，会导致IO异常，单次批量插入上限2100条左右；需要对数组逐步截取，这里选择的是1000条为一批，每1000条数据落库一次

//单次分批落库条数
int subList = 1000;
//计算执行次数
int count = personList.size() % subList > 0 ? (personList.size()/subList)+1 : personList.size()/subList;
//临时集合
List<Person> tempList;
for (int i = 0; i < count; i++) {//截取集合tempList = personList.stream().skip(subList*i).limit(subList).collect(Collectors.toList());//数据落库personDao.insertBatch(personList);
}

mybatis中，使用批量插入的方式落库，可以提高插入的效率

<insert id="insertBatch" parameterType="com.example.pojo.Person">insert into person(id,identity,name,age,address,desc) values<foreach collection="personList" item="item" index="index" separator="," >(#{item.id,jdbcType=VARCHAR},#{item.identity,jdbcType=VARCHAR},#{item.name,jdbcType=VARCHAR},#{item.age,jdbcType=INTEGER},#{item.address,jdbcType=VARCHAR},#{item.desc,jdbcType=VARCHAR})</foreach>
</insert>

4.多线程方式落库

目标：实现异步落库

使用springboot线程池，多线程可以提高落库效率，for循环走的很快，需要使用CountDownLatch中await方法，等待全部线程执行完毕再结束

//单次分批落库条数
int subList = 1000;
//计算执行次数
int count = personList.size() % subList > 0 ? (personList.size()/subList)+1 : personList.size()/subList;
//临时集合
List<Person> tempList;
CountDownLatch countDownLatch = new CountDownLatch(personList.size()/subList);
for (int i = 0; i < count; i++) {//截取集合tempList = personList.stream().skip(subList*i).limit(subList).collect(Collectors.toList());//多线程执行落库方法asyncService.executeAsync(tempList, countDownLatch);
}
//阻塞线程，等待全部线程执行完毕
countDownLatch.await();

springboot线程池代码

@Service
@EnableAsync
public class AsyncService {@Autowiredprivate PersonDao personDao;@Async("taskExecutor")public void executeAsync(List<Person> personList, CountDownLatch countDownLatch) {try{personDao.insertBatch(personList);} finally {countDownLatch.countDown();}}
}

@Configuration
@EnableAsync
public class ExecutorConfig {ThreadPoolProperties properties = new ThreadPoolProperties();@Bean(name = "taskExecutor")public ThreadPoolTaskExecutor taskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(properties.getCorePoolSize());executor.setMaxPoolSize(properties.getMaxPoolSize());executor.setQueueCapacity(properties.getQueueCapacity());executor.setThreadNamePrefix(properties.getThreadNamePrefix());executor.setKeepAliveSeconds(properties.getKeepAliveTime());executor.setWaitForTasksToCompleteOnShutdown(properties.isWaitForTasksToCompleteOnShutdown());executor.setAwaitTerminationSeconds(properties.getAwaitTerminationSeconds());executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());return executor;}@Dataclass ThreadPoolProperties {private int corePoolSize = 30;private int maxPoolSize = 100;private int keepAliveTime;private int queueCapacity = 1000;private String threadNamePrefix = "test";private boolean allowCoreThreadTimeout = false;private boolean waitForTasksToCompleteOnShutdown = false;private int awaitTerminationSeconds;}
}

二、总结

通过打时间戳记录时间，读取文件到存储实体集合用时10秒，时间主要花费在落库；单线程总用时18分钟左右，多线程4分钟；还可以将表的所有字段取消主键约束和非空约束提高落库效率，最终优化后总用时仅32秒！

原创不易，转载请注明来源

SpringBoot线程池实现200w数据快速落库相关推荐

Springboot线程池的使用和扩展
我们常用ThreadPoolExecutor提供的线程池服务,springboot框架提供了@Async注解,帮助我们更方便的将业务逻辑提交到线程池中异步执行,今天我们就来实战体验这个线程池服务: 实 ...
SpringBoot 线程池，也太好用了叭！
欢迎关注方志朋的博客,回复"666"获面试宝典来源:blog.csdn.net/m0_37701381/article/details/81072774 前言前两天做项目的时候 ...
SpringBoot线程池的创建、@Async配置步骤及注意事项
点击上方"方志朋",选择"设为星标" 回复"666"获取新整理的面试资料作者:Muscleheng blog.csdn.net/Muscl ...
SpringBoot线程池ThreadPoolTaskExecutor和@Async异步方法浅解及代码应用示例
目录线程池执行顺序线程池配置策略 Spring线程池的配置类: Spring有.无返回值的异步调用示例自定义的异步方法类代码: 测试类代码常见问题: 参考文章: 线程池执行顺序核心线程数(C ...
JAVA使用线程池查询大批量数据
前言在开发过程中可能会碰到某些独特的业务,比如查询全部表数据,数据量过多会导致查询变得十分缓慢.虽然在大多数情况下并不需要查询所有的数据,而是通过分页或缓存的形式去减少或者避免这个问题,但是仍然存在 ...
springboot 线程池_Spring boot 2 线程池怎么配置
线程池在java 中线程池,就是ThreadPoolExecutor来构造,简单看下线程池包含的方法, corePoolSize:初始化线程.线程池中保留的线程数量. maximumPoolSize ...
springboot线程池配置
1. application.yml配置 # 异步线程配置 # 配置核心线程数 async.executor.thread.core_pool_size: 10 # 配置最大线程数 async.exe ...
flink实时流遇到的问题排查——部分数据未落库redis问题
flink实时流遇到的问题排查 1.技术和环境 2.问题表述 3.简化的代码 4.问题排查思路 5.结论 6.后续补充 1.技术和环境技术:kafka.zookeeper.DataStream.re ...
springboot线程池使用
1.配置@Configuration @EnableAsync public class TaskPoolConfig {@Bean("taskExecutor")public E ...
Springboot应用中线程池配置教程（2021版）
前言:日常开发中我们常用ThreadPoolExecutor提供的线程池服务帮我们管理线程,在Springboot中更是提供了@Async注解来简化业务逻辑提交到线程池中执行的过程.由于Springb ...

SpringBoot线程池实现200w数据快速落库