JVM堆溢出解决方法:Java大数据迁移实战
本文针对Java微服务在大数据迁移(如百万级记录)时易发的JVM堆内存溢出问题,提供了一套基于数据库分页的批处理解决方案。文章深入剖析了内存溢出的原因,并详细阐述了如何利用数据库的LIMIT和OFFSET特性,结合应用程序的迭代处理逻辑,实现数据的分批查询与处理,有效降低单次操作的内存消耗,避免服务崩溃。通过优化SQL查询构建方法和改造数据归档流程,保证数据迁移的稳定性和效率,为Java微服务处理海量数据提供了一种实用的实践方案。
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
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 应用程序层面的迭代处理
在应用程序中,我们需要构建一个循环,在每次迭代中:
- 计算当前的 OFFSET 值。
- 使用 LIMIT 和计算出的 OFFSET 从源数据库中查询一个批次的数据。
- 处理这个批次的数据(例如,复制到目标表)。
- 更新 OFFSET 值,为下一次迭代做准备。
- 当查询结果为空时,表示所有数据已处理完毕,退出循环。
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<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) {
今天带大家了解了的相关知识,希望对你有所帮助;关于文章的技术知识我们会一点点深入介绍,欢迎大家关注golang学习网公众号,一起学习编程~

- 上一篇
- GolangHTTP分块上传优化技巧

- 下一篇
- Golang微服务版本兼容全攻略
-
- 文章 · java教程 | 10分钟前 |
- RecyclerView数据跨Adapter传递技巧
- 417浏览 收藏
-
- 文章 · java教程 | 19分钟前 |
- Android通知渠道优先级与优先级区别解析
- 433浏览 收藏
-
- 文章 · java教程 | 24分钟前 |
- JavaStreamAPI过滤映射排序全解析
- 477浏览 收藏
-
- 文章 · java教程 | 32分钟前 |
- SpringBoot日志配置与异步优化方法
- 216浏览 收藏
-
- 文章 · java教程 | 43分钟前 |
- Java实战Spark处理气象大数据全解析
- 103浏览 收藏
-
- 文章 · java教程 | 49分钟前 |
- SpringCloud微服务注册中心搭建指南
- 304浏览 收藏
-
- 文章 · java教程 | 55分钟前 |
- JavaLambda与Stream使用详解
- 149浏览 收藏
-
- 文章 · java教程 | 1小时前 |
- Java生成高级Excel报表教程
- 399浏览 收藏
-
- 文章 · java教程 | 1小时前 |
- SpringBoot整合RocketMQ事务消息教程
- 293浏览 收藏
-
- 文章 · java教程 | 1小时前 | 内存泄漏 内存碎片 Java堆外内存 DirectByteBuffer Unsafe
- Java堆外内存优化技巧分享
- 279浏览 收藏
-
- 前端进阶之JavaScript设计模式
- 设计模式是开发人员在软件开发过程中面临一般问题时的解决方案,代表了最佳的实践。本课程的主打内容包括JS常见设计模式以及具体应用场景,打造一站式知识长龙服务,适合有JS基础的同学学习。
- 542次学习
-
- GO语言核心编程课程
- 本课程采用真实案例,全面具体可落地,从理论到实践,一步一步将GO核心编程技术、编程思想、底层实现融会贯通,使学习者贴近时代脉搏,做IT互联网时代的弄潮儿。
- 509次学习
-
- 简单聊聊mysql8与网络通信
- 如有问题加微信:Le-studyg;在课程中,我们将首先介绍MySQL8的新特性,包括性能优化、安全增强、新数据类型等,帮助学生快速熟悉MySQL8的最新功能。接着,我们将深入解析MySQL的网络通信机制,包括协议、连接管理、数据传输等,让
- 497次学习
-
- JavaScript正则表达式基础与实战
- 在任何一门编程语言中,正则表达式,都是一项重要的知识,它提供了高效的字符串匹配与捕获机制,可以极大的简化程序设计。
- 487次学习
-
- 从零制作响应式网站—Grid布局
- 本系列教程将展示从零制作一个假想的网络科技公司官网,分为导航,轮播,关于我们,成功案例,服务流程,团队介绍,数据部分,公司动态,底部信息等内容区块。网站整体采用CSSGrid布局,支持响应式,有流畅过渡和展现动画。
- 484次学习
-
- 边界AI平台
- 探索AI边界平台,领先的智能AI对话、写作与画图生成工具。高效便捷,满足多样化需求。立即体验!
- 394次使用
-
- 免费AI认证证书
- 科大讯飞AI大学堂推出免费大模型工程师认证,助力您掌握AI技能,提升职场竞争力。体系化学习,实战项目,权威认证,助您成为企业级大模型应用人才。
- 405次使用
-
- 茅茅虫AIGC检测
- 茅茅虫AIGC检测,湖南茅茅虫科技有限公司倾力打造,运用NLP技术精准识别AI生成文本,提供论文、专著等学术文本的AIGC检测服务。支持多种格式,生成可视化报告,保障您的学术诚信和内容质量。
- 542次使用
-
- 赛林匹克平台(Challympics)
- 探索赛林匹克平台Challympics,一个聚焦人工智能、算力算法、量子计算等前沿技术的赛事聚合平台。连接产学研用,助力科技创新与产业升级。
- 641次使用
-
- 笔格AIPPT
- SEO 笔格AIPPT是135编辑器推出的AI智能PPT制作平台,依托DeepSeek大模型,实现智能大纲生成、一键PPT生成、AI文字优化、图像生成等功能。免费试用,提升PPT制作效率,适用于商务演示、教育培训等多种场景。
- 549次使用
-
- 提升Java功能开发效率的有力工具:微服务架构
- 2023-10-06 501浏览
-
- 掌握Java海康SDK二次开发的必备技巧
- 2023-10-01 501浏览
-
- 如何使用java实现桶排序算法
- 2023-10-03 501浏览
-
- Java开发实战经验:如何优化开发逻辑
- 2023-10-31 501浏览
-
- 如何使用Java中的Math.max()方法比较两个数的大小?
- 2023-11-18 501浏览