【文件增量备份系统】使用Mysql的流式查询优化数据清理性能（针对百万量级数据）

本文主要是介绍【文件增量备份系统】使用Mysql的流式查询优化数据清理性能（针对百万量级数据），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

功能介绍
原始方案
- 测试
流式处理
- 测试
功能可用性测试

功能介绍

清理功能的作用是：扫描数据库中已经备份过的文件，查看数据源中是否还有相应的文件，如果没有，说明该文件被删除了，那相应的，也需要将备份目标目录的文件以及相关的备份记录都一并删除掉

原始方案

使用分批处理，避免单次加载表中的所有数据，导致发现内存溢出，每次从备份文件表中查询出2000条备份文件记录，然后对查出来的数据进行检验、清理

/*** 检查数据，删除 无效备份信息 和 已备份文件* 什么叫无效？简单来说就是，已备份文件和原文件对应不上，或者说原文件被删除了** @param sourceId*/
@Override
public void clearBySourceIdv1(Long sourceId) {long current = 1;ClearTask clearTask = new ClearTask();clearTask.setId(snowFlakeUtil.nextId());// 填充数据源相关信息BackupSource source = backupSourceService.getById(sourceId);if (source == null) {throw new ClientException("所需要清理的数据源不存在");}clearTask.setClearSourceRoot(source.getRootPath());// 存储要删除的文件List<Long> removeBackupFileIdList = new ArrayList<>();List<String> removeBackupTargetFilePathList = new ArrayList<>();BackupFileRequest backupFileRequest = new BackupFileRequest();backupFileRequest.setBackupSourceId(sourceId);backupFileRequest.setSize(2000L);long totalFileNum = -1;long finishFileNum = 0;ClearStatistic clearStatistic = new ClearStatistic(0);while (true) { 查询数据，监测看哪些文件需要被删除// 分页查询出数据，即分批检查，避免数据量太大，占用太多内存backupFileRequest.setCurrent(current);PageResponse<BackupFile> backupFilePageResponse = backupFileService.pageBackupFile(backupFileRequest);if (totalFileNum == -1 && backupFilePageResponse.getTotal() != null) {totalFileNum = backupFilePageResponse.getTotal();Map<String, Object> dataMap = new HashMap<>();dataMap.put("code", WebsocketNoticeEnum.CLEAR_START.getCode());dataMap.put("message", WebsocketNoticeEnum.CLEAR_START.getDetail());clearTask.setTotalFileNum(totalFileNum);clearTask.setFinishFileNum(0L);clearTask.setClearStatus(0);clearTask.setClearNumProgress("0.0");clearTask.setStartTime(new DateTime());clearTask.setClearTime(0L);dataMap.put("clearTask", clearTask);webSocketServer.sendMessage(JSON.toJSONString(dataMap), WebSocketServer.usernameAndSessionMap.get("Admin"));}if (backupFilePageResponse.getRecords().size() > 0) {for (BackupFile backupFile : backupFilePageResponse.getRecords()) {// 获取备份文件的路径// todo 待优化为存储的时候，不存储整一个路径，节省数据库空间，只存储从根目录开始后面的路径，后面获取整个路径再进行拼接String sourceFilePath = backupFile.getSourceFilePath();File sourceFile = new File(sourceFilePath);if (!sourceFile.exists()) {// --if-- 如果原目录该文件已经被删除，则删除removeBackupFileIdList.add(backupFile.getId());removeBackupTargetFilePathList.add(backupFile.getTargetFilePath());}}// 换一页来检查current += 1;} else {// 查不出数据了，说明检查完了break;} 执行删除if (removeBackupFileIdList.size() > 0) {// 批量删除无效备份文件backupFileService.removeByIds(removeBackupFileIdList);// 删除无效的已备份文件for (String backupTargetFilePath : removeBackupTargetFilePathList) {File removeFile = new File(backupTargetFilePath);if (removeFile.exists()) {boolean delete = FileUtils.recursionDeleteFiles(removeFile, clearStatistic);if (!delete) {throw new ServiceException("文件无法删除");}}}// 批量删除无效备份文件对应的备份记录backupFileHistoryService.removeByFileIds(removeBackupFileIdList);removeBackupFileIdList.clear();removeBackupTargetFilePathList.clear();}// 告诉前端，更新清理状态finishFileNum += backupFilePageResponse.getRecords().size();Map<String, Object> dataMap = new HashMap<>();dataMap.put("code", WebsocketNoticeEnum.CLEAR_PROCESS.getCode());dataMap.put("message", WebsocketNoticeEnum.CLEAR_PROCESS.getDetail());clearTask.setFinishFileNum(finishFileNum);clearTask.setClearStatus(1);clearTask.setFinishDeleteFileNum(clearStatistic.finishDeleteFileNum);setClearProgress(clearTask, dataMap);}// 清理成功Map<String, Object> dataMap = new HashMap<>();dataMap.put("code", WebsocketNoticeEnum.CLEAR_SUCCESS.getCode());dataMap.put("message", WebsocketNoticeEnum.CLEAR_SUCCESS.getDetail());clearTask.setFinishFileNum(finishFileNum);clearTask.setClearStatus(2);clearTask.setFinishDeleteFileNum(clearStatistic.finishDeleteFileNum);setClearProgress(clearTask, dataMap);dataMap.put("clearTask", clearTask);
}

测试

经过测试，发现该方案非常慢，清理进度10%竟要花费3分钟

在这里插入图片描述

通过观察，发现备份文件数量一共有接近三百多万条，如此大的数据量，使用分页查询的性能会非常差。这是因为每次分页查询，都需要从头开始扫描，若分页的页码越大，分页查询的速度也会越慢

在这里插入图片描述

流式处理

流式处理方式即使用数据库的流式查询功能，查询成功之后不是返回一个数据集合，而是返回一个迭代器，通过这个迭代器可以进行循环，每次查询出一条数据来进行处理。使用该方式可以有效降低内存占用，且因为不需要像分页一样每次重头扫描表，每查询一条数据都是在上次查询的基础上面查询，即知道上条数据的位置，因此查询效率较高

/*** 流式处理* 检查数据，删除 无效备份信息 和 已备份文件* 什么叫无效？简单来说就是，已备份文件和原文件对应不上，或者说原文件被删除了** @param sourceId*/
@SneakyThrows
public void clearBySourceIdV2(Long sourceId) {// 获取 dataSource Bean 的连接@Cleanup Connection conn = dataSource.getConnection();@Cleanup Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);stmt.setFetchSize(Integer.MIN_VALUE);long start = System.currentTimeMillis();// 查询sql，只查询关键的字段String sql = "SELECT id,source_file_path,target_file_path FROM backup_file where backup_source_id = " + sourceId;@Cleanup ResultSet rs = stmt.executeQuery(sql);loopResultSetProcessClear(rs, sourceId);log.info("流式清理花费时间：{} s ", (System.currentTimeMillis() - start) / 1000);
}/*** 循环读取，每次读取一行数据进行处理** @param rs* @param sourceId* @return*/
@SneakyThrows
private Long loopResultSetProcessClear(ResultSet rs, Long sourceId) {// 填充数据源相关信息BackupSource source = backupSourceService.getById(sourceId);if (source == null) {throw new ClientException("所需要清理的数据源不存在");}// 中途用来存储需要删除的文件信息List<Long> removeBackupFileIdList = new ArrayList<>();List<String> removeBackupTargetFilePathList = new ArrayList<>();// 查询文件总数long totalFileNum = backupFileService.count(Wrappers.query(new BackupFile()).eq("backup_source_id", sourceId));// 已经扫描的文件数量long finishFileNum = 0;ClearStatistic clearStatistic = new ClearStatistic(0);long second = System.currentTimeMillis() / 1000;long curSecond;// 发送消息通知前端 清理正式开始ClearTask clearTask = ClearTask.builder().id(snowFlakeUtil.nextId()).clearSourceRoot(source.getRootPath()).totalFileNum(totalFileNum).finishFileNum(0L).clearStatus(0).clearNumProgress("0.0").startTime(new DateTime()).clearTime(0L).build();Map<String, Object> dataMap = new HashMap<>();dataMap.put("clearTask", clearTask);notify(WebsocketNoticeEnum.CLEAR_START, dataMap);// 每次获取一行数据进行处理，rs.next()如果有数据返回true，否则返回falsewhile (rs.next()) {// 获取数据中的属性long fileId = rs.getLong("id");String sourceFilePath = rs.getString("source_file_path");String targetFilePath = rs.getString("target_file_path");// 所扫描的文件数量+1finishFileNum++;// 获取备份文件的路径File sourceFile = new File(sourceFilePath);if (!sourceFile.exists()) {// --if-- 如果原目录该文件已经被删除，则删除removeBackupFileIdList.add(fileId);removeBackupTargetFilePathList.add(targetFilePath);}if (removeBackupFileIdList.size() >= 2000) {clear(removeBackupFileIdList, removeBackupTargetFilePathList, clearStatistic);}curSecond = System.currentTimeMillis() / 1000;if (curSecond > second) {second = curSecond;// 告诉前端，更新清理状态clearTask.setFinishFileNum(finishFileNum);clearTask.setClearStatus(1);clearTask.setFinishDeleteFileNum(clearStatistic.finishDeleteFileNum);setClearProgress(clearTask, dataMap);notify(WebsocketNoticeEnum.CLEAR_PROCESS, dataMap);}}// 循环结束之后，再清理一次，避免文件数没有到达清理批量导致清理失败clear(removeBackupFileIdList, removeBackupTargetFilePathList, clearStatistic);// 告诉前端，清理成功clearTask.setFinishFileNum(finishFileNum);clearTask.setClearStatus(2);clearTask.setFinishDeleteFileNum(clearStatistic.finishDeleteFileNum);setClearProgress(clearTask, dataMap);notify(WebsocketNoticeEnum.CLEAR_SUCCESS, dataMap);return 0L;
}/*** 执行清理* @param removeBackupFileIdList* @param removeBackupTargetFilePathList* @param clearStatistic*/
private void clear(List<Long> removeBackupFileIdList, List<String> removeBackupTargetFilePathList, ClearStatistic clearStatistic) {// 批量删除无效备份文件backupFileService.removeByIds(removeBackupFileIdList);// 删除无效的已备份文件for (String backupTargetFilePath : removeBackupTargetFilePathList) {File removeFile = new File(backupTargetFilePath);if (removeFile.exists()) {boolean delete = FileUtils.recursionDeleteFiles(removeFile, clearStatistic);if (!delete) {throw new ServiceException("文件无法删除");}}}// 批量删除无效备份文件对应的备份记录backupFileHistoryService.removeByFileIds(removeBackupFileIdList);removeBackupFileIdList.clear();removeBackupTargetFilePathList.clear();
}/*** 发送通知给前端** @param noticeEnum* @param dataMap*/
private void notify(WebsocketNoticeEnum noticeEnum, Map<String, Object> dataMap) {dataMap.put("code", noticeEnum.getCode());dataMap.put("message", noticeEnum.getDetail());webSocketServer.sendMessage(JSON.toJSONString(dataMap), WebSocketServer.usernameAndSessionMap.get("Admin"));
}