JVM堆溢出解决方法：Java大数据迁移实战

本文针对Java微服务在大数据迁移（如百万级记录）时易发的JVM堆内存溢出问题，提供了一套基于数据库分页的批处理解决方案。文章深入剖析了内存溢出的原因，并详细阐述了如何利用数据库的LIMIT和OFFSET特性，结合应用程序的迭代处理逻辑，实现数据的分批查询与处理，有效降低单次操作的内存消耗，避免服务崩溃。通过优化SQL查询构建方法和改造数据归档流程，保证数据迁移的稳定性和效率，为Java微服务处理海量数据提供了一种实用的实践方案。

本文旨在解决Java微服务在处理海量数据（如百万级记录迁移）时遇到的JVM堆内存耗尽问题。通过深入探讨内存溢出的根源，并提出一种高效的批处理策略，利用数据库的LIMIT和OFFSET机制分批次查询和处理数据，从而显著降低单次操作的内存消耗，避免服务崩溃，确保数据迁移的稳定性和效率。

1. 问题背景与内存溢出分析

在Java微服务架构中，当需要处理或迁移大量数据（例如百万级数据库记录）时，常见的做法是查询所有符合条件的数据并加载到内存中进行后续处理。然而，这种一次性加载海量数据的策略极易导致JVM堆内存耗尽（java.lang.OutOfMemoryError: Java heap space），表现为服务崩溃或响应缓慢。

错误日志通常会显示类似以下信息：

Resource exhaustion event: the JVM was unable to allocate memory from the heap.
ResourceExhausted! (1/0)

这表明应用程序试图分配的内存超出了JVM堆的可用容量。在数据迁移场景中，如将一个表中的大量记录复制到另一个表，如果一次性查询并缓存所有源数据（例如使用 JdbcTemplate.queryForList()），即使后续使用了JDBC批处理进行写入，内存压力也可能在数据查询阶段就达到极限。

原始代码片段中的 List> sourceRecords = sourceDbTemplate.queryForList(...) 是导致内存问题的核心。当 sourceRecords 包含数百万条记录时，每条记录作为一个 Map 对象，连同其内部的键值对，将迅速消耗大量堆内存。

2. 解决方案：基于数据库分页的批处理策略

为了解决一次性加载海量数据导致的内存溢出问题，核心思想是将大批量数据处理分解为多个小批次处理。这可以通过结合数据库的分页查询能力和应用程序的迭代处理逻辑来实现。

2.1 数据库层面的批次查询 (LIMIT 和 OFFSET)

数据库提供了 LIMIT（或 TOP）和 OFFSET（或 SKIP）子句，允许我们指定查询结果的数量以及从哪个位置开始返回。这是实现数据分批查询的基础。

SQL查询示例：

SELECT *
FROM your_table
WHERE your_condition
ORDER BY unique_id_column -- 必须指定一个排序字段，确保每次查询的顺序稳定
LIMIT batch_size
OFFSET current_offset;

• LIMIT batch_size: 定义每个批次要查询的记录数量。
• OFFSET current_offset: 定义从结果集的哪个位置开始返回记录。current_offset 会随着已处理记录的数量递增。
• ORDER BY unique_id_column: 至关重要！ 必须根据一个稳定、唯一且通常是索引的列（如主键ID、创建时间戳等）进行排序。这确保了：
  • 每次分页查询的结果顺序是确定的。
  • 不会遗漏或重复获取数据。
  • 对于大型数据集和高 OFFSET 值，数据库可能需要扫描大量跳过的行，因此 OFFSET 的性能会随之下降。在极端情况下，可以考虑使用基于游标（Keyset Pagination）的方法（即 WHERE id > last_id ORDER BY id LIMIT batch_size），但这超出了本教程的范围。

2.2 应用程序层面的迭代处理

在应用程序中，我们需要构建一个循环，在每次迭代中：

1. 计算当前的 OFFSET 值。
2. 使用 LIMIT 和计算出的 OFFSET 从源数据库中查询一个批次的数据。
3. 处理这个批次的数据（例如，复制到目标表）。
4. 更新 OFFSET 值，为下一次迭代做准备。
5. 当查询结果为空时，表示所有数据已处理完毕，退出循环。

3. 示例代码实现

下面我们将基于原有的代码结构，展示如何修改 archiveTableRecords 和 buildSQLQueryToFetchSourceRecords 方法以实现批处理。

3.1 修改SQL查询构建方法

为了支持 LIMIT 和 OFFSET，我们需要修改 ArchiveSQLQueries.buildSQLQueryToFetchSourceRecords 方法，使其能够接收批次大小、偏移量以及用于排序的列名。

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.ParameterizedPreparedStatementSetter;

// 假设 ArchiveConfigDTO 和 getCSTDateNew 方法已定义
// 假设 ArchiveSQLQueries 是一个包含静态方法的类

public class DataArchiverService {

    private static final Logger logger = LoggerFactory.getLogger(DataArchiverService.class);

    @Value("${batch-size}")
    private int batchSize; // 配置的批处理大小，例如1000或5000

    // 辅助方法，假设存在
    private String getCSTDateNew(String archiveMonths) {
        // 实现日期计算逻辑
        return "2022-09-01"; // 示例值
    }

    // 辅助方法，假设存在
    private int getSumOfArray(int[][] array) {
        int sum = 0;
        for (int[] innerArray : array) {
            for (int value : innerArray) {
                sum += value;
            }
        }
        return sum;
    }

    // --- 修改后的 ArchiveSQLQueries 类中的方法 ---
    public static class ArchiveSQLQueries {
        public static String buildSQLQueryToFetchSourceRecords(String sourceTable, String orderByColumn, int limit, int offset) {
            StringBuilder sb = new StringBuilder("SELECT * FROM " + sourceTable + " where update_dts <= ?");
            // 确保 orderByColumn 是安全的，避免SQL注入风险
            if (orderByColumn != null && !orderByColumn.isEmpty()) {
                sb.append(" ORDER BY ").append(orderByColumn);
            }
            sb.append(" LIMIT ").append(limit);
            sb.append(" OFFSET ").append(offset);
            return sb.toString();
        }

        public static String buildSQLTargetRecordInsertionQuery(String targetTable, Map record, String primaryKeyColumn) {
            // 假设此方法已正确实现，根据Map构建INSERT语句
            // 示例：INSERT INTO target_table (col1, col2) VALUES (?, ?)
            StringBuilder sb = new StringBuilder("INSERT INTO ").append(targetTable).append(" (");
            StringBuilder values = new StringBuilder(" VALUES (");

            boolean firstColumn = true;
            for (String key : record.keySet()) {
                if (!key.equals(primaryKeyColumn)) { // 假设主键在插入时是自增的，不包含在VALUES中
                    if (!firstColumn) {
                        sb.append(", ");
                        values.append(", ");
                    }
                    sb.append(key);
                    values.append("?");
                    firstColumn = false;
                }
            }
            sb.append(")").append(values).append(")");
            return sb.toString();
        }
    }

    // --- 原始的 copySourceRecords 方法，其内部已使用批处理写入 ---
    public int copySourceRecords(JdbcTemplate targetDbTemplate, String targetTable, String primaryKeyColumn,
                                 List> sourceRecords, List