MyBatis流式查询实战：解决大数据量查询OOM问题

这次我们来看一个 Java 开发中非常实际的问题：如何用 MyBatis 的流式查询，优雅地解决大数据量查询导致的内存溢出（OOM）。如果你遇到过查询几十万、上百万条数据时，程序直接卡死或抛出OutOfMemoryError的情况，那么这篇文章就是为你准备的。

MyBatis 的流式查询（Streaming Query）并不是一个新概念，但很多开发者对其理解不深，或者知道但不敢用、不会用。它的核心价值在于，它允许你像处理水流一样处理数据库查询结果，边读边处理，而不是一次性把所有数据都加载到 JVM 内存里。这对于报表导出、数据同步、ETL 处理等场景是救命稻草。

本文将直接切入主题，不讲复杂的理论，重点放在“能不能用”和“怎么用”上。我们会先快速了解流式查询的核心能力与使用边界，然后通过一个 Spring Boot 项目，从零开始演示如何配置、启动和测试流式查询。你会看到如何用一行代码触发内存危机，再用另一行代码（实际上是正确的配置和调用方式）化解危机。我们重点关注其工作原理、资源占用（内存和游标）、接口调用方式，以及如何集成到批量任务中。最后，会给出完整的常见问题排查清单和最佳实践建议。

无论你是正在面试准备“MyBatis 如何防止 OOM”这类八股文，还是在实际开发中遇到了性能瓶颈，这篇文章都能提供可直接落地的解决方案。

1. 核心能力速览

在深入代码之前，我们先通过一个表格快速把握 MyBatis 流式查询的全貌，明确它能做什么、有什么要求。

2. 适用场景与使用边界

理解了核心能力，我们再来明确一下流式查询的用武之地和注意事项。

最适合它的战场：

1. 数据导出：这是最经典的场景。用户点击“导出全部数据”，后台可能需要查询百万行记录并生成 Excel 或 CSV。用传统方式，内存瞬间爆炸。流式查询可以边读边写文件，内存曲线几乎是一条直线。
2. 数据同步与迁移：需要将 A 数据库的表数据全量读取，处理后写入 B 数据库或消息队列。流式查询可以作为一个稳定的“生产者”，平稳地输出数据流。
3. 批量计算与统计：需要对海量数据进行逐条或分批的复杂计算（如风控规则匹配、用户画像更新），且计算逻辑不适合在 SQL 中完成。流式查询允许你将数据“流”进计算引擎。
4. 日志处理：处理应用日志表，进行归档、分析和清理。

需要谨慎或避免使用的场景：

1. 需要反复遍历结果集：流式游标只能向前（TYPE_FORWARD_ONLY），不能回头。如果你需要多次使用同一份数据，应该先用传统方式加载到内存（如果数据量允许），或者考虑其他方案。
2. 结果集很小：对于几千条记录，一次性加载的耗时和内存开销完全可以接受。引入流式查询反而增加了代码复杂性和事务管理成本，得不偿失。
3. 高并发短事务业务：流式查询会长时间占用一个数据库连接（直到游标关闭）。在并发量极高的 OLTP 场景中，这可能成为连接池的瓶颈。它更适合后台任务、离线处理等对响应时间不敏感的场景。

合规与安全边界：

• 数据安全：流式处理的是敏感数据时，仍需确保输出通道（如文件、消息队列）的安全，防止数据泄露。
• 资源管理：必须确保游标被正确关闭，否则会导致数据库连接泄漏，最终拖垮整个应用。这是开发者的首要责任。
• 超时控制：长时间运行的流式查询需要设置合理的语句执行超时和事务超时，避免“僵尸”查询占用资源。

3. 环境准备与前置条件

接下来，我们搭建一个最小化的测试环境来验证流式查询。你将需要以下准备：

1. Java 开发环境：

   • JDK 8 或更高版本（推荐 JDK 11 或 17）。本文示例基于 JDK 17。
   • Maven 3.6+ 或 Gradle 作为构建工具。
   • 一个 IDE，如 IntelliJ IDEA 或 Eclipse。

2. Spring Boot 项目骨架：

   • 使用 Spring Initializr 快速生成一个项目。
   • 依赖选择：Spring Web,Spring Data JDBC(或MyBatis Framework)，MySQL Driver(根据你的数据库选择)。
   • 本文示例将使用mybatis-spring-boot-starter。

3. 数据库：

   • 一个可用的 MySQL 实例（版本 5.7+ 或 8.0+）。其他数据库如 PostgreSQL 原理类似，主要区别在于 JDBC 驱动和部分连接参数。
   • 创建一张有足够多数据的测试表。我们可以用一段简单的 SQL 脚本快速生成百万级测试数据。

4. 关键依赖 (Mavenpom.xml)： 确保你的pom.xml中包含 MyBatis Spring Boot Starter 和数据库驱动。

   <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!-- MyBatis 集成 --> <dependency> <groupId>org.mybatis.spring.boot</groupId> <artifactId>mybatis-spring-boot-starter</artifactId> <version>3.0.3</version> <!-- 请使用最新稳定版 --> </dependency> <!-- MySQL 驱动 --> <dependency> <groupId>com.mysql</groupId> <artifactId>mysql-connector-j</artifactId> <scope>runtime</scope> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> </dependencies>

5. 数据库表准备： 执行以下 SQL 创建表和生成测试数据。

   -- 创建测试表 CREATE TABLE `large_data_table` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `user_name` varchar(255) DEFAULT NULL, `email` varchar(255) DEFAULT NULL, `created_at` datetime DEFAULT CURRENT_TIMESTAMP, `some_data` text, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 使用存储过程快速生成 100 万条测试数据（根据机器性能，可能需要几分钟） DELIMITER $$ CREATE PROCEDURE generate_test_data() BEGIN DECLARE i INT DEFAULT 1; WHILE i <= 1000000 DO INSERT INTO `large_data_table` (`user_name`, `email`, `some_data`) VALUES ( CONCAT('user_', i), CONCAT('user_', i, '@example.com'), REPEAT(CONCAT('Sample data for row ', i, ' '), 10) -- 每条记录约 200+ 字符 ); SET i = i + 1; END WHILE; END$$ DELIMITER ; -- 执行存储过程 CALL generate_test_data(); -- 执行完毕后，可以删除存储过程 DROP PROCEDURE generate_test_data;

4. 安装部署与启动方式

环境准备好后，我们开始编写代码。这里没有复杂的“安装部署”，核心是 MyBatis 的配置和代码编写。

1. 配置文件 (application.yml)： 在src/main/resources/application.yml中配置数据库连接和 MyBatis 的基本设置。关键点在于为流式查询配置fetchSize。

   spring: datasource: url: jdbc:mysql://localhost:3306/your_database?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true # 对于流式查询，这个参数至关重要！它告诉JDBC驱动每次从网络流中抓取多少行。 # 设置为 Integer.MIN_VALUE 是 MySQL 驱动识别流式结果集的一个特殊值。 # 也可以设置为一个正整数，如 1000，表示每次抓取1000行。 connection-properties: useCursorFetch=true;defaultFetchSize=-2147483648 username: your_username password: your_password driver-class-name: com.mysql.cj.jdbc.Driver hikari: # 连接池配置，根据实际情况调整 maximum-pool-size: 10 mybatis: # 指定 mapper.xml 文件位置，如果使用注解则非必须 mapper-locations: classpath:mapper/*.xml configuration: # 开启驼峰命名映射 map-underscore-to-camel-case: true # 日志实现，方便调试 log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

   注意：defaultFetchSize=-2147483648即Integer.MIN_VALUE，这是让 MySQL JDBC 驱动启用流式结果集的标志。useCursorFetch=true也是必要的参数。

2. 实体类 (LargeData.java)： 对应数据库表的实体类。

   package com.example.demo.entity; import java.time.LocalDateTime; public class LargeData { private Long id; private String userName; private String email; private LocalDateTime createdAt; private String someData; // 省略 getter, setter, toString 方法 }

3. Mapper 接口 (LargeDataMapper.java)： 这是核心。我们定义两个方法，一个返回List（用于对比），一个返回Cursor。

   package com.example.demo.mapper; import com.example.demo.entity.LargeData; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Select; import org.apache.ibatis.cursor.Cursor; import java.util.List; @Mapper public interface LargeDataMapper { // 传统方式：一次性加载所有数据到 List - “内存杀手” @Select("SELECT id, user_name, email, created_at, some_data FROM large_data_table") List<LargeData> selectAllAsList(); // 流式查询：返回 Cursor，数据不会一次性加载到内存 @Select("SELECT id, user_name, email, created_at, some_data FROM large_data_table") Cursor<LargeData> selectAllAsCursor(); }

4. Service 层 (DataProcessService.java)： 在这里实现业务逻辑。流式查询必须在事务内执行！

   package com.example.demo.service; import com.example.demo.entity.LargeData; import com.example.demo.mapper.LargeDataMapper; import lombok.extern.slf4j.Slf4j; import org.apache.ibatis.cursor.Cursor; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import java.io.BufferedWriter; import java.io.FileWriter; import java.io.IOException; @Service @Slf4j public class DataProcessService { private final LargeDataMapper largeDataMapper; public DataProcessService(LargeDataMapper largeDataMapper) { this.largeDataMapper = largeDataMapper; } /** * 危险操作：一次性加载到List，数据量大时必OOM */ public void processWithList() { log.info("开始使用List方式查询..."); long start = System.currentTimeMillis(); // 这一行代码，就是潜在的“内存挤爆器” var list = largeDataMapper.selectAllAsList(); log.info("查询完成，共 {} 条记录", list.size()); // 模拟处理 for (LargeData data : list) { // do something with data } long end = System.currentTimeMillis(); log.info("List方式处理完成，耗时: {} ms", (end - start)); } /** * 安全操作：使用流式查询，边读边处理 * @Transactional 注解确保整个游标操作在一个数据库事务中 */ @Transactional public void processWithCursor() { log.info("开始使用Cursor流式查询..."); long start = System.currentTimeMillis(); long count = 0; // 使用 try-with-resources 确保 Cursor 被自动关闭 try (Cursor<LargeData> cursor = largeDataMapper.selectAllAsCursor()) { for (LargeData data : cursor) { count++; // 在这里处理每一条数据，例如写入文件、发送到消息队列、进行计算等 // 模拟处理 if (count % 10000 == 0) { log.info("已处理 {} 条记录", count); } } } // 此处自动调用 cursor.close() long end = System.currentTimeMillis(); log.info("Cursor流式处理完成，共处理 {} 条记录，耗时: {} ms", count, (end - start)); } /** * 更实际的例子：流式查询并导出到CSV文件 */ @Transactional public void exportToCsv(String filePath) throws IOException { try (Cursor<LargeData> cursor = largeDataMapper.selectAllAsCursor(); BufferedWriter writer = new BufferedWriter(new FileWriter(filePath))) { // 写入CSV头 writer.write("id,user_name,email,created_at"); writer.newLine(); for (LargeData data : cursor) { // 拼接一行数据 String line = String.format("%d,%s,%s,%s", data.getId(), data.getUserName(), data.getEmail(), data.getCreatedAt()); writer.write(line); writer.newLine(); } log.info("数据已导出至: {}", filePath); } // 自动关闭 cursor 和 writer } }

5. 启动与测试控制器 (DemoController.java)： 创建一个简单的 REST 端点来触发我们的测试。

   package com.example.demo.controller; import com.example.demo.service.DataProcessService; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import java.io.IOException; @RestController @RequestMapping("/api/demo") public class DemoController { private final DataProcessService dataProcessService; public DemoController(DataProcessService dataProcessService) { this.dataProcessService = dataProcessService; } @GetMapping("/oom") public String triggerOom() { // 警告：这个接口很可能导致应用OOM崩溃，仅供演示，生产环境切勿暴露！ dataProcessService.processWithList(); return "传统List查询完成（如果还没OOM的话）"; } @GetMapping("/stream") public String triggerStream() { dataProcessService.processWithCursor(); return "流式查询处理完成"; } @GetMapping("/export") public String triggerExport() throws IOException { dataProcessService.exportToCsv("./exported_data.csv"); return "流式导出完成，文件保存在项目根目录"; } }

6. 启动应用： 运行 Spring Boot 主类（通常是DemoApplication.java），应用启动后，访问http://localhost:8080/api/demo/stream即可测试流式查询。

5. 功能测试与效果验证

现在，让我们通过实际调用来验证两种方式的巨大差异。

测试 1：传统 List 查询（模拟 OOM 场景）

目的：直观感受一次性加载百万数据对内存的冲击。操作：

1. 启动应用，确保 JVM 最大堆内存（Xmx）设置得较小，例如-Xmx256m，以便快速触发 OOM。
2. 访问http://localhost:8080/api/demo/oom。预期结果：

• 应用日志会显示开始查询，然后很快停滞。
• 控制台大概率会抛出java.lang.OutOfMemoryError: Java heap space错误。
• 应用可能无响应或崩溃。判断成功：成功触发 OOM 或观察到内存监控曲线（如 JVisualVM, Arthas）中堆内存瞬间飙升到顶。（这个“成功”恰恰证明了传统方式的危险）

测试 2：Cursor 流式查询

目的：验证流式查询的稳定性和低内存占用。操作：

1. 重启应用（如果上一步崩溃了）。
2. 访问http://localhost:8080/api/demo/stream。预期结果：

• 应用日志平稳输出 “已处理 10000 条记录”、“已处理 20000 条记录”……直到结束。
• 通过内存监控工具观察，会发现堆内存使用率有轻微、平稳的波动，但绝不会出现陡峭的峰值。
• 最终成功输出处理完成的日志，应用运行正常。判断成功：完整处理 100 万条记录，且应用内存平稳，未发生 OOM。

测试 3：流式查询导出到文件

目的：验证流式查询在真实业务场景（数据导出）中的应用。操作：

1. 访问http://localhost:8080/api/demo/export。
2. 观察项目根目录下是否生成了exported_data.csv文件。预期结果：

• 文件被成功创建，并且大小随着处理进度逐渐增大。
• 处理过程中，应用内存占用依然平稳。
• 最终得到一个包含 100 万行数据的 CSV 文件。判断成功：文件生成，内容完整，内存无异常。

实测观察要点：

• CPU 和 I/O：流式查询时，CPU 和磁盘 I/O（如果是导出到文件）会成为主要瓶颈，而不是内存。
• 数据库负载：数据库服务器需要维持一个长时间的游标，可能会占用一些资源。对于 MySQL，可以观察SHOW PROCESSLIST中对应连接的状态。
• 网络流量：数据是分批传输的，网络流量也是平稳的。

6. 接口 API 与批量任务集成

流式查询本身是数据访问层技术，但它可以完美地作为后端 API 或批量任务的核心引擎。

作为 REST API 返回流式响应

对于数据导出 API，我们可以直接返回一个流式响应，让客户端（如浏览器）边下载边接收数据，进一步提升体验。

import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.transaction.annotation.Transactional; import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody; @GetMapping(value = "/download/csv", produces = "text/csv") @Transactional // 事务仍然必要！ public ResponseEntity<StreamingResponseBody> downloadCsv() { String filename = "data_export.csv"; // 设置响应头，告诉浏览器这是文件下载 HttpHeaders headers = new HttpHeaders(); headers.add(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"" + filename + "\""); headers.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_OCTET_STREAM_VALUE); StreamingResponseBody stream = outputStream -> { try (Cursor<LargeData> cursor = largeDataMapper.selectAllAsCursor(); PrintWriter writer = new PrintWriter(new OutputStreamWriter(outputStream, StandardCharsets.UTF_8))) { writer.write("id,user_name,email,created_at\n"); for (LargeData data : cursor) { writer.write(String.format("%d,%s,%s,%s\n", data.getId(), data.getUserName(), data.getEmail(), data.getCreatedAt())); writer.flush(); // 及时刷新缓冲区，实现流式输出 } } }; return ResponseEntity.ok().headers(headers).body(stream); }

这样，前端调用这个接口，就会立即开始下载文件，服务器端是边查边写，内存压力极小。

集成到批量任务框架（如 Spring Batch, Quartz）

在定时任务或批处理作业中，流式查询可以作为ItemReader来使用。

import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.support.AbstractItemStreamItemReader; import org.apache.ibatis.cursor.Cursor; import org.apache.ibatis.session.SqlSessionFactory; public class MyBatisCursorItemReader<T> extends AbstractItemStreamItemReader<T> { private final SqlSessionFactory sqlSessionFactory; private final String queryId; private Cursor<T> cursor; public MyBatisCursorItemReader(SqlSessionFactory sqlSessionFactory, String queryId) { this.sqlSessionFactory = sqlSessionFactory; this.queryId = queryId; setSaveState(false); // 游标状态通常不保存 } @Override @Transactional public T read() throws Exception { if (cursor == null) { // 开启游标 cursor = sqlSessionFactory.openSession().selectCursor(queryId); } if (cursor != null && cursor.hasNext()) { return cursor.next(); } else { // 处理完毕，关闭资源 if (cursor != null) { cursor.close(); } return null; } } @Override public void close() { super.close(); if (cursor != null) { cursor.close(); } } }

在 Spring Batch 的配置中，将这个 Reader 注入到 Step 中，就可以高效、安全地处理海量数据了。

7. 资源占用与性能观察

理解流式查询的资源占用模式，对于调优和排错至关重要。

内存占用观察：

• 传统 List 方式：JVM 堆内存使用量会随着ResultSet全部加载到内存而直线上升，峰值接近(单条记录大小) * (记录条数)。这是导致 OOM 的直接原因。
• 流式 Cursor 方式：堆内存使用量保持在一个较低且稳定的水平。内存中通常只保留fetchSize指定数量的记录（MySQL 驱动在流式模式下可能会忽略fetchSize或使用最小值）。你可以使用 JVisualVM、JConsole 或 Arthas 的dashboard命令观察Heap Memory Usage曲线，对比两种方式的差异。

数据库连接与游标资源：

• 流式查询会长时间占用一个数据库连接，直到Cursor被关闭或遍历结束。这意味着：
  1. 连接池中的这个连接在该事务期间不能被其他线程使用。
  2. 必须确保在 finally 块或 try-with-resources 中关闭游标，否则会导致连接泄漏。
• 在数据库端（如 MySQL），可以通过SHOW PROCESSLIST看到该连接处于Sending data状态。游标资源在数据库服务器端也会被占用，直到客户端关闭它。

性能权衡：

• 总耗时：对于网络状况良好、数据量巨大的情况，流式查询的总耗时可能略高于一次性查询。因为多了多次网络往返的开销。但这个时间差换来了系统的稳定性，是可接受的。
• 响应时间：对于需要即时响应的 API，流式查询的首次结果返回速度可能更快（因为不需要等待所有数据都取回），适合分页或“懒加载”场景，但 MyBatis Cursor 本身并不直接支持分页，它是一次性遍历。

监控建议：

1. 监控应用服务器的堆内存使用率，确保流式处理时曲线平稳。
2. 监控数据库的活跃连接数和长时间运行的查询。
3. 在业务日志中记录流式处理的开始、进度和结束，便于追踪。

8. 常见问题与排查方法

在实际使用流式查询时，你可能会遇到以下问题。这里提供一份排查清单。

9. 最佳实践与使用建议

为了在生产环境中稳定、高效地使用 MyBatis 流式查询，请遵循以下建议：

1. 事务边界要清晰：将@Transactional注解加在调用流式查询的服务层方法上，确保整个遍历过程在一个事务内。不要在遍历中途做会导致事务提交或回滚的操作。
2. 务必关闭游标：使用try-with-resources语法是关闭Cursor的最安全、最简洁的方式。绝对不要在遍历后忘记关闭它。
3. 保持处理逻辑轻量：循环体内的业务逻辑应尽可能高效。如果单条处理很慢，百万条数据的总时间会非常长。考虑将耗时的操作（如网络调用、复杂计算）异步化或批量处理。
4. 合理设置超时：在数据库连接字符串和连接池中设置合理的超时时间（如socketTimeout,connectionTimeout），防止网络问题导致线程永久阻塞。
5. 监控与告警：对使用流式查询的任务进行监控，记录其开始时间、处理条数、结束时间和状态。设置告警，如果任务运行时间异常长，及时通知开发人员。
6. 做好兜底和重试：流式处理长时间任务时，可能因网络抖动、数据库维护等中断。设计任务时考虑断点续传或幂等性，以便在失败后能从断点恢复。
7. 区分使用场景：再次强调，不是所有查询都需要流式。对于小数据量、需要多次访问结果集、或需要高并发响应的场景，请坚持使用传统的List方式。
8. 进行性能测试：在上线前，使用和生产环境类似的数据量进行充分的性能测试，评估流式查询对数据库和应用的负载影响，找到最优的fetchSize等参数。

MyBatis 流式查询是一个强大的工具，它用相对简单的 API 解决了大数据量处理中的核心内存难题。关键在于理解其“边读边处理”的核心理念和“事务内操作、必须关闭”的约束。通过本文的步骤，你可以在自己的项目中快速集成并验证这一能力。

下次当你面对“导出全部数据”或“批量处理全表”的需求时，不必再为 OOM 提心吊胆。正确配置，加上一个Cursor，就能让数据像溪流一样平稳地流过你的系统，而不是像洪水一样瞬间冲垮内存堤坝。建议将本文中的配置示例和排查清单收藏备用，在遇到相关问题时能快速定位。