批量数据持久化时内存泄漏问题排查

核心问题 #

你的代码最大的问题是一次性将所有CSV数据加载到内存并持有引用至整个事务结束，导致GC无法及时回收这些大对象：

• c1/c2/c3三个大集合（总计252万条数据）在transactionTemplate.execute的lambda作用域内被全程持有，从加载完成到整个事务结束（删除+全量持久化完成），这些对象一直有强引用，GC根本没机会回收。
• 事务范围过大：加载CSV、删除数据、持久化数据全在一个事务里，进一步拉长了大对象的生命周期。

另外，虽然BulkObjectRepository里调用了entityManager.clear()清理了持久化上下文，但这只释放了EntityManager缓存的实体，原始的CSV转换后的对象列表（c1/c2/c3）还在内存里被引用着，所以GC无法回收。

具体修复步骤 #

1. 分阶段处理，避免同时持有所有大集合

不要一次性加载所有CSV，而是处理完一个表的所有操作（加载→删除→持久化）后，再处理下一个，让前一个集合的引用失效，GC可以及时回收：

fun importData() {
    logger.info("Import started")
    
    // 分表处理，每个表的操作单独在事务内执行
    transactionTemplate.execute {
        logger.info("Loading Csv1...")
        val c1 = importCsv1()
        logger.info("Deleting C1 Table...")
        c1Repository.deleteAllInBatch()
        logger.info("Persisting C1...")
        bulkObjectRepository.persist(c1)
    }

    transactionTemplate.execute {
        logger.info("Loading Csv2...")
        val c2 = importCsv2()
        logger.info("Deleting C2 Table...")
        c2Repository.deleteAllInBatch()
        logger.info("Persisting C2...")
        bulkObjectRepository.persist(c2)
    }

    transactionTemplate.execute {
        logger.info("Loading Csv3...")
        val c3 = importCsv3()
        logger.info("Deleting C3 Table...")
        c3Repository.deleteAllInBatch()
        logger.info("Persisting C3...")
        bulkObjectRepository.persist(c3)
        logger.info("Import finished")
    }

    //do some stuff here that must be outside transaction
}

2. 流式处理CSV（最优方案）

如果CSV支持流式读取，不要一次性将所有数据转成对象列表，而是边读边批量持久化，彻底避免内存中持有百万级别的对象集合：
修改importCsvX方法，返回Sequence<Any>或者直接在读取时就调用批量持久化，比如：

// 示例：流式读取CSV并批量持久化
fun importAndPersistCsv3() {
    transactionTemplate.execute {
        c3Repository.deleteAllInBatch()
        val csvReader = openCsvStreamReader() // 流式读取的CSV阅读器
        var batch = mutableListOf<Any>()
        for (csvRow in csvReader) {
            val entity = convertToEntity(csvRow)
            batch.add(entity)
            if (batch.size >= 5000) {
                bulkObjectRepository.persistBatch(batch) // 批量持久化当前批次
                batch.clear()
            }
        }
        if (batch.isNotEmpty()) {
            bulkObjectRepository.persistBatch(batch)
        }
    }
}

对应的BulkObjectRepository新增批量方法：

fun persistBatch(entities: List<Any>) {
    entities.forEach { entityManager.persist(it) }
    entityManager.flush()
    entityManager.clear()
}

3. 缩小事务范围（可选）

将CSV加载过程移出事务，事务只包含数据库操作（删除+持久化），避免事务上下文额外占用内存：

fun importData() {
    logger.info("Import started")
    
    // 先加载数据（事务外）
    logger.info("Loading Csv1...")
    val c1 = importCsv1()
    logger.info("Loading Csv2...")
    val c2 = importCsv2()
    logger.info("Loading Csv3...")
    val c3 = importCsv3()

    // 数据库操作放在事务内，执行完后及时切断引用
    transactionTemplate.execute {
        logger.info("Deleting Tables...")
        c1Repository.deleteAllInBatch()
        c2Repository.deleteAllInBatch()
        c3Repository.deleteAllInBatch()

        logger.info("Persisting C1...")
        bulkObjectRepository.persist(c1)
        logger.info("Persisting C2...")
        bulkObjectRepository.persist(c2)
        logger.info("Persisting C3...")
        bulkObjectRepository.persist(c3)
        logger.info("Import finished")
    }

    //do some stuff here that must be outside transaction
}

为什么手动GC能释放内存？ #

手动执行jcmd GC.run时，JVM会强制触发全量GC，此时即使这些大对象还在作用域内（如果是在事务执行完之后），或者JVM判断没有后续引用，就会回收它们。但正常情况下，只有当这些对象的强引用被移除后，GC才会在自动回收周期里处理它们。

内容的提问来源于stack exchange，提问作者ochs.tobi