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

作者：你好赵伟 | 2024-03-13 05:28:01

踩

文章目录

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

功能介绍

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

原始方案

使用分批处理，避免单次加载表中的所有数据，导致发现内存溢出，每次从备份文件表中查询出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);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

测试

经过测试，发现该方案非常慢，清理进度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，否则返回false
    while (rs.next()) {
        // 获取数据中的属性
        long fileId = rs.getLong("id");
        String sourceFilePath = rs.getString("source_file_path");
        String targetFilePath = rs.getString("target_file_path");

        // 所扫描的文件数量+1
        finishFileNum++;

        // 获取备份文件的路径
        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"));
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146