一.需求分析
最近接到一个需求,导入十万级,甚至可能百万数据量的记录了车辆黑名单的Excel文件,借此机会分析下编码过程;
首先将这个需求拆解,发现有三个比较复杂的问题:
问题一:Excel文件导入后首先要被解析为存放对象的列表,数据量大的情况下可能会导致内存溢出,解析时间过长;
问题二:插入数据库的时候,数据量大,写入的时间长
问题三:要对数据库中的现有数据进项判断,不仅仅要做插入动作,还要将数据库的数据与导入的数据对比,判断是否做更新操作
其中:
问题一和问题三,可以看做同一类,因为主要涉及内存计算导致的性能问题,以及内存占用过大的溢出问题,
关于这两个问题,现在线上的机器基本上是4核8G的配置集群部署,内存并不是关键,我会在另一篇文章中给出我的方案,
今天主要针对问题二,写入的数据库的问题给出我的方案,
问题二主要是多次写入数据库的问题,显然,如果有几十万条数据,那么是不可能连续写几十万次的,不然要写到后年马月才能全部入库,
解决方案:
这里我主要采用了多线程的写入方式,十万条数据,2000条写一次(可以自己定义),用线程池提交多个线程任务同时写入,提高性能
二.代码环境
Springboot2.1.3+POI+PGSQL
controller层代码
@PostMapping("/upload")
public void upload1(MultipartFile file, @Validated UploadReq req) throws Exception {
//从数据库查询出现有的数据,根据去重的字段分组去构建成一个HashMap,通过containsKey()判断
//将需要更新的数据放到updateList中
List<User> updateList=new ArrayList<>();
//已取值的行数
int rowNum = 0;
//列号
int colNum = 0;
//真正有数据的行数
int realRowCount = 0;
//得到工作空间
Workbook workbook = null;
try {
workbook = ExcelUtil.getWorkbookByInputStream(file.getInputStream(), file.getOriginalFilename());
} catch (IOException e) {
e.printStackTrace();
}
//得到工作表
int numberOfSheets = workbook.getNumberOfSheets();
for (int i = 0; i < numberOfSheets; i++) {
Sheet sheet = ExcelUtil.getSheetByWorkbook(workbook, i)
realRowCount = sheet.getPhysicalNumberOfRows();
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
List<User> list = new ArrayList<>();
User user = null;
for(Row row:sheet) {
if(realRowCount == rowNum) {
break;
}
//空行跳过
if(ExcelUtil.isBlankRow(row)) {
continue;
}
if(row.getRowNum() == -1) {
continue;
}else {
//第一行表头跳过
if(row.getRowNum() == 0) {
continue;
}
}
rowNum ++;
colNum = 1;
user = new User();
ExcelUtil.validCellValue(sheet, row, colNum, "id");
user.setId(Integer.valueOf(ExcelUtil.getCellValue(sheet, row, colNum - 1)));
ExcelUtil.validCellValue(sheet, row, ++ colNum, "name");
user.setId(Integer.valueOf(ExcelUtil.getCellValue(sheet, row, colNum - 1)));
//判断是否是已存在的数据,如果是就更新,不是就新增
//updateList.add(user);
list.add(user);
}
//新增的逻辑
userService.saveBatch(list);
System.out.println(list);
}
}
service层代码
@Service public class UserServiceImpl implements IUserService { @Autowired private UserMapper userMapper; @Override public void saveBatch(List<User> list) throws Exception { //一个线程处理200条数据 int count = 200; //数据集合大小 int listSize = list.size(); //开启的线程数 int runSize = (listSize / count) + 1; //存放每个线程的执行数据 List<User> newlist = null; //创建一个线程池，数量和开启线程的数量一样 //Executors 的写法 // ExecutorService executor = Executors.newFixedThreadPool(runSize); //ThreadPoolExecutor的写法 ThreadPoolExecutor executor = new ThreadPoolExecutor(runSize, runSize, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3), new ThreadPoolExecutor.DiscardOldestPolicy()); //创建两个个计数器 CountDownLatch begin = new CountDownLatch(1); CountDownLatch end = new CountDownLatch(runSize); //循环创建线程 for (int i = 0; i < runSize; i++) { //计算每个线程执行的数据 if ((i + 1) == runSize) { int startIndex = (i * count); int endIndex = list.size(); newlist = list.subList(startIndex, endIndex); } else { int startIndex = (i * count); int endIndex = (i + 1) * count; newlist = list.subList(startIndex, endIndex); } //线程类 ImportThread mythead = new ImportThread(newlist, begin, end,userMapper); //这里执行线程的方式是调用线程池里的executor.execute(mythead)方法。 executor.execute(mythead); } begin.countDown(); end.await(); //执行完关闭线程池 executor.shutdown(); }
线程类
public class ImportThread implements Runnable { public ImportThread() { } UserMapper userMapper; private List<User> list; private CountDownLatch begin; private CountDownLatch end; /** * 方法名: ImportThread * 方法描述: 创建个构造函数初始化 list,和其他用到的参数 * @throws */ public ImportThread(List<User> list, CountDownLatch begin, CountDownLatch end,UserMapper userMapper) { this.list = list; this.begin = begin; this.end = end; this.userMapper=userMapper; } @Override public void run() { try { //执行完让线程直接进入等待 userMapper.saveBatch(list); begin.await(); } catch (InterruptedException e) { e.printStackTrace(); } finally { //这里要主要了，当一个线程执行完 了计数要减一不然这个线程会被一直挂起 //这个方法就是直接把计数器减一的 end.countDown(); } } }