Java微服务中高效处理海量数据：避免JVM内存溢出的分批策略

在Java微服务中处理百万级甚至千万级的数据记录时，常见的“Resource exhaustion event: the JVM was unable to allocate memory from the heap”错误通常源于一次性将所有数据加载到内存中。尽管可能使用 batchUpdate 进行批量写入，但如果数据源的读取本身没有分批，JVM依然会因为持有大量数据对象而耗尽内存。解决此问题的核心在于将数据处理流程分解为“分批读取”和“分批处理”两个阶段。

核心策略：分批数据读取与处理

为了避免JVM内存溢出，我们必须改变一次性查询所有数据的做法，转而采用迭代式的分批查询。这主要通过数据库的 LIMIT 和 OFFSET 子句实现，每次只查询固定数量的记录，处理完成后再查询下一批。

1. 分批查询 (Batch Fetching): 使用 LIMIT 和 OFFSET SQL子句来限制每次查询返回的记录数量。LIMIT 指定返回的最大记录数，OFFSET 指定从结果集的哪一行开始返回。

   SELECT *
   FROM your_table
   WHERE your_condition
   ORDER BY unique_id_column -- 必须有ORDER BY确保每次分页结果一致
   LIMIT batch_size
   OFFSET current_offset;

2. 确保结果一致性 (Consistency with ORDER BY): 在进行分页查询时，ORDER BY 子句至关重要。它确保每次查询的数据顺序是确定的，从而避免在不同批次中出现重复记录或遗漏记录。通常，选择一个唯一且有序的列（如主键ID、创建时间戳等）作为排序依据。如果主键是自增ID，它是非常理想的选择。

3. 迭代处理 (Iterative Processing): 在一个循环中重复执行分批查询，每次查询后更新 OFFSET 值，直到不再有数据返回。

实现细节：基于JdbcTemplate的分批查询与处理

以下是基于Spring JdbcTemplate 实现分批数据抓取和处理的伪代码及示例：

首先，定义一个配置项来控制每批处理的数据量：

立即学习“Java免费学习笔记（深入）”；

@Value("${data.batch-fetch-size:10000}") // 默认每次抓取10000条记录
private int batchFetchSize;

接下来，修改数据抓取和处理的主逻辑，使其能够迭代地处理数据：

public void archiveTableRecords(JdbcTemplate sourceDbTemplate, JdbcTemplate targetDbTemplate,
                                ArchiveConfigDTO archiveObj) {
    try {
        String sourceTable = archiveObj.getSourceTable();
        String archive_months = archiveObj.getArchiveCriteriaMonths();
        String primaryKeyColumn = archiveObj.getPrimaryKeyColumn(); // 假设主键列名
        String compareDate = getCSTDateNew(archive_months);
        logger.info("Archive criteria date: {}", compareDate);

        int processedRecords = 0;
        List> sourceRecords;

        do {
            // 1. 分批查询数据
            String fetchSql = ArchiveSQLQueries.buildSQLQueryToFetchSourceRecordsBatched(
                                    sourceTable, primaryKeyColumn, processedRecords, batchFetchSize);
            sourceRecords = sourceDbTemplate.queryForList(fetchSql, compareDate);

            if (!sourceRecords.isEmpty()) {
                logger.info("Fetched {} {} record(s) from offset {}", sourceRecords.size(), sourceTable, processedRecords);

                // 2. 批量处理（复制和删除）
                List