百万级数据场景下优化月度预算账户读写插入效率的方案问询

当前处理1000条记录耗时32149秒，数据库存有百万级记录，现有代码性能瓶颈明显。

现有代码 #

/**
 * Saves monthly budget accounts to the database.
 * 
 * @param dto                       the list of monthly budget accounts to save
 * @param branchRepo                the repository for branches
 * @param costCodesRepo             the repository for cost codes
 * @param budgetVersionRepo         the repository for budget versions
 * @param monthlyBudgetAccountsRepo the repository for monthly budget accounts
 * @return the number of accounts saved
 */
public int saveMonthlyBudgetAccounts(List<MonthlyBudgetAccountsDTO> dto) {

    long startTime = System.currentTimeMillis(); 
    
    // Fetch required data from database and store in memory for later use
    List<Branch> branchList = branchRepo.findAll();
    Map<String, Branch> branchMap = branchList.parallelStream()
            .collect(Collectors.toMap(b -> b.getSapBranch().toLowerCase(), Function.identity()));

    List<CostCodes> costCodesList = costCodesRepo.findAll();
    Map<String, CostCodes> costCodesMap = costCodesList.parallelStream()
            .collect(Collectors.toMap(c -> c.getCostCodeDescription().toUpperCase().trim(), Function.identity()));

    List<BudgetVersion> budgetVersionList = budgetVersionRepo.findAll();
    Map<String, BudgetVersion> budgetVersionMap = budgetVersionList.parallelStream()
            .collect(Collectors.toMap(b -> b.getBudgetVersion().toLowerCase().trim(), Function.identity()));

    int count = 0;
    
    
    for (MonthlyBudgetAccountsDTO acc : dto) {
        MonthlyBudgetAccounts m = new MonthlyBudgetAccounts();
        MonthlyBudgetAccountsPK pk = new MonthlyBudgetAccountsPK();
        String branch = acc.getId().getBranch().toLowerCase();
        String costCode = acc.getId().getCostCode().toUpperCase();
        String budgetVersion = acc.getId().getBudgetVersion().toLowerCase();
        try {
            // Set the branch
            Branch branchObj = branchMap.getOrDefault(branch.substring(branch.length() - 4), null);
            if (branchObj == null) {
                // Handle case where branch is not found
                branchObj = branchRepo.findById(900); // Replace with default branch object
            }
            pk.setBranch(branchObj);

            // Set the cost code
            CostCodes costCodeObj = costCodesMap.get(costCode.trim());
            if (costCodeObj == null) {
                // Handle case where cost code is not found
                throw new NoSuchElementException("Cost code not found");
            }
            pk.setCostCode(costCodeObj);

            // Set the budget version
            BudgetVersion budgetVersionObj = budgetVersionMap.get(budgetVersion.trim());
            if (budgetVersionObj == null) {
                // Handle case where budget version is not found
                throw new NoSuchElementException("Budget version not found");
            }
            pk.setBudgetVersion(budgetVersionObj);

            // Set other fields
            pk.setAccount(acc.getId().getAccount());
            pk.setFinancialYear(acc.getId().getFinancialYear());

            m.setId(pk);
            m.setTotal(acc.getTotal());
            m.setMarch(acc.getMarch());
            m.setApril(acc.getApril());
            m.setMay(acc.getMay());
            m.setJune(acc.getJune());
            m.setJuly(acc.getJuly());
            m.setAugust(acc.getAugust());
            m.setSeptember(acc.getSeptember());
            m.setOctober(acc.getOctober());
            m.setNovember(acc.getNovember());
            m.setDecember(acc.getDecember());
            m.setJanuary(acc.getJanuary());
            m.setFebruary(acc.getFebruary());
            m.setInsertDate(acc.getInsertDate().trim());

            monthlyBudgetAccountsRepo.save(m);

            // Clear the DTO to free up memory
            dto.set(count, null);
            count++;

            if (count % 1000 == 0) {
                System.out.println(count + " records processed");
                long endTime = System.currentTimeMillis();
                System.out.println(count + " - Time Taken: " + (endTime - startTime) + "ms");
                startTime = 0;
            }
        } catch (NoSuchElementException e) {
            // Handle any errors that occur while processing the record
            e.printStackTrace();
        }
    }
    return count;
}

优化方案 #

1. 批量插入替代单条保存 #

这是最核心的优化点：原代码每次循环调用save(m)会触发单次数据库IO，百万级数据下IO开销会被无限放大。改成批量保存：

• 维护临时列表，每积累500-2000条（可根据数据库配置调整）就调用saveAll()批量插入
• 示例代码片段：

List<MonthlyBudgetAccounts> batchList = new ArrayList<>(1000);
int count = 0;
long startTime = System.currentTimeMillis();

for (MonthlyBudgetAccountsDTO acc : dto) {
    // ... 构建MonthlyBudgetAccounts对象的逻辑不变 ...
    
    batchList.add(m);
    count++;

    if (batchList.size() % 1000 == 0) {
        monthlyBudgetAccountsRepo.saveAll(batchList);
        batchList.clear();
        // 打印耗时
        long endTime = System.currentTimeMillis();
        System.out.println(count + " records processed, Time Taken: " + (endTime - startTime) + "ms");
        startTime = endTime;
    }
}
// 处理剩余未提交的数据
if (!batchList.isEmpty()) {
    monthlyBudgetAccountsRepo.saveAll(batchList);
}

2. 并行流与映射优化 #

• 若findAll()返回的数据量较小，并行流的线程切换开销可能超过收益，可根据实际数据量选择是否使用并行流
• Collectors.toMap遇到重复key会抛出异常，建议添加合并函数避免报错：

Map<String, Branch> branchMap = branchList.stream()
        .collect(Collectors.toMap(
            b -> b.getSapBranch().toLowerCase(), 
            Function.identity(), 
            (oldVal, newVal) -> oldVal // 重复key时保留旧值
        ));

3. 减少重复数据库查询 #

• 原代码中分支找不到时，每次循环都调用branchRepo.findById(900)，可提前缓存默认分支到内存：

// 初始化阶段提前查询默认分支
Branch defaultBranch = branchRepo.findById(900);

// 循环内直接使用缓存的默认分支
Branch branchObj = branchMap.getOrDefault(branch.substring(branch.length() - 4), defaultBranch);

4. 内存与GC优化 #

• 原代码中dto.set(count, null)操作无意义，反而可能破坏列表结构，处理完的DTO会自动被GC回收，无需手动置空
• 避免在循环内重复创建对象（如MonthlyBudgetAccounts、MonthlyBudgetAccountsPK），可考虑对象池复用（注意线程安全）

5. 数据库层面优化 #

• 给Branch.sapBranch、CostCodes.costCodeDescription、BudgetVersion.budgetVersion这些查询字段建立索引，提升数据读取与映射构建效率
• 手动控制事务：用大事务包裹所有批量插入操作，减少事务提交开销（注意事务过大可能导致锁表，需权衡批次大小）
• 调整数据库连接池参数，增大最大连接数，避免因连接不足导致线程等待

内容的提问来源于stack exchange，提问作者DynComplier