在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<Map<String, Object>> sourceRecords = sourceDbTemplate.queryForList(...) 是导致内存问题的核心。当 sourceRecords 包含数百万条记录时，每条记录作为一个 Map<String, Object> 对象，连同其内部的键值对，将迅速消耗大量堆内存。
立即学习“Java免费学习笔记（深入）”；
为了解决一次性加载海量数据导致的内存溢出问题，核心思想是将大批量数据处理分解为多个小批次处理。这可以通过结合数据库的分页查询能力和应用程序的迭代处理逻辑来实现。
数据库提供了 LIMIT（或 TOP）和 OFFSET（或 SKIP）子句，允许我们指定查询结果的数量以及从哪个位置开始返回。这是实现数据分批查询的基础。
SQL查询示例：
SELECT * FROM your_table WHERE your_condition ORDER BY unique_id_column -- 必须指定一个排序字段，确保每次查询的顺序稳定 LIMIT batch_size OFFSET current_offset;
在应用程序中，我们需要构建一个循环，在每次迭代中：
下面我们将基于原有的代码结构，展示如何修改 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<String, Object> 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<Map<String, Object>> sourceRecords, List<Object> primaryKeyValueList) {
int result = 0;
logger.info("Copying {} records to {}", sourceRecords.size(), targetTable);
if (sourceRecords.isEmpty()) {
return 0;
}
// 构建插入语句，基于第一个记录的结构
String insertSql = ArchiveSQLQueries.buildSQLTargetRecordInsertionQuery(targetTable, sourceRecords.get(0), primaryKeyColumn);
int[][] insertResult = targetDbTemplate.batchUpdate(
insertSql,
sourceRecords,
batchSize, // 这里使用了配置的batchSize进行JDBC批处理写入
new ParameterizedPreparedStatementSetter<Map<String, Object>>() {
@Override
public void setValues(PreparedStatement ps, Map<String, Object> argument) throws SQLException {
int index = 1;
for (Entry<String, Object> obj : argument.entrySet()) {
// 假设主键列在目标表是自增的，或者不作为插入参数
if (obj.getKey().equals(primaryKeyColumn)) {
primaryKeyValueList.add(obj.getValue()); // 收集主键值
} else {
ps.setObject(index++, obj.getValue());
}
}
}
});
result = getSumOfArray(insertResult);
logger.info("Inserted {} record(s) in {}", result, targetTable);
return result;
}
// --- 修改后的 archiveTableRecords 方法，实现批次循环 ---
public void archiveTableRecords(JdbcTemplate sourceDbTemplate, JdbcTemplate targetDbTemplate,
ArchiveConfigDTO archiveObj) {
try {
String sourceTable = archiveObj.getSourceTable();
String targetTable = archiveObj.getTargetTable();
String primaryKeyColumn = archiveObj.getPrimaryKeyColumn(); // 假设这是用于排序的列
String archive_months = archiveObj.getArchiveCriteriaMonths();
String compareDate1 = getCSTDateNew(archive_months);
logger.info("Archive criteria date: {}", compareDate1);
int processedRecords = 0;
List<Object> allPrimaryKeyValueList = new ArrayList<>(); // 用于收集所有已处理记录的主键，以便后续删除
while (true) {
// 1. 批次查询源数据
List<Map<String, Object>> sourceRecordsBatch = sourceDbTemplate.queryForList(
ArchiveSQLQueries.buildSQLQueryToFetchSourceRecords(sourceTable, primaryKeyColumn, batchSize, processedRecords),
compareDate1
);
if (sourceRecordsBatch.isEmpty()) {
logger.info("No more records to fetch for table {}. Total processed: {}", sourceTable, processedRecords);
break; // 没有更多记录，退出循环
}
logger.info("Fetched batch of {} records from {} (offset: {})", sourceRecordsBatch.size(), sourceTable, processedRecords);
// 2. 准备当前批次的主键列表
List<Object> currentBatchPrimaryKeys = new ArrayList<>();
// 3. 复制当前批次数据到目标表
int recordsInsertedInBatch = copySourceRecords(targetDbTemplate, targetTable, primaryKeyColumn, sourceRecordsBatch, currentBatchPrimaryKeys);
if (recordsInsertedInBatch > 0) {
// 将当前批次的主键添加到总列表中
allPrimaryKeyValueList.addAll(currentBatchPrimaryKeys);
logger.info("Copied {} record(s) to {}. Total copied: {}", recordsInsertedInBatch, targetTable, allPrimaryKeyValueList.size());
}
// 4. 更新已处理记录数，作为下一次查询的偏移量
processedRecords += sourceRecordsBatch.size();
}
// 5. 所有批次处理完毕后，根据收集到的主键删除源数据
if (!allPrimaryKeyValueList.isEmpty()) {
logger.info("Initiating deletion of {} records from source table {}.", allPrimaryKeyValueList.size(), sourceTable);
// 注意：如果 allPrimaryKeyValueList 极其庞大，deleteSourceRecords 方法也可能需要内部批处理
deleteSourceRecords(sourceDbTemplate, sourceTable, primaryKeyColumn, allPrimaryKeyValueList);
logger.info("Deletion completed for {} records from source table {}.", allPrimaryKeyValueList.size(), sourceTable);
}
} catch (Exception e) {
logger.error("Exception in archiveTableRecords: {} {}", e.getMessage(), e);
// 在实际应用中，这里需要更健壮的异常处理，例如记录失败的批次信息，以便后续重试
}
}
// 假设 deleteSourceRecords 方法已存在
public void deleteSourceRecords(JdbcTemplate sourceDbTemplate, String sourceTable, String primaryKeyColumn, List<Object> primaryKeyValueList) {
// 示例：DELETE FROM source_table WHERE primaryKeyColumn IN (?, ?, ...)
// 对于非常大的 primaryKeyValueList，需要考虑分批次执行DELETE语句，或使用其他高效的删除策略
String deleteSql = "DELETE FROM " + sourceTable + " WHERE " + primaryKeyColumn + " IN (" +
String.join(",", java.util.Collections.nCopies(primaryKeyValueList.size(), "?")) + ")";
try {
sourceDbTemplate.batchUpdate(deleteSql, new ArrayList<>(primaryKeyValueList)); // 假设可以一次性处理
} catch (Exception e) {以上就是解决JVM堆内存溢出：Java应用高效处理海量数据批次迁移实践的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
广告
收藏
收藏
收藏
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
536
