你有没有遇到过这样的场景一个看似简单的数据导出功能在测试环境跑得好好的一到生产环境就突然内存飙升直接 OOMOutOfMemoryError把服务干趴下你查了半天日志发现罪魁祸首只是一条平平无奇的SELECT * FROM large_table。这不是段子而是很多后端开发者踩过的真实大坑。当数据量从几千条变成几百万条时传统的 JDBC 查询方式会瞬间将海量数据全部加载到 JVM 内存中内存被“挤爆”几乎是必然结果。很多人第一反应是调大 JVM 堆内存但这只是饮鸩止渴数据再大一点呢今天要聊的MyBatis 流式查询就是专门用来“拆弹”的。它不是什么高深的新技术却是处理大数据量查询时避免内存溢出的“标准答案”。很多人知道这个概念但在实际项目中要么不敢用要么用错了反而引入了连接泄露的新问题。本文将彻底讲清楚为什么一行普通的查询代码能“挤爆”内存MyBatis 流式查询的原理是什么如何正确、安全地使用它以及在什么场景下你其实根本不需要它。1. 这篇文章真正要解决的问题我们首先得达成一个共识技术方案的选择永远是对“成本”的权衡。这里的成本包括内存成本、CPU成本、网络I/O成本以及最重要的——开发和维护的复杂度成本。核心问题当你的 Java 应用需要从数据库查询并处理大量数据比如百万级以上时如何避免一次性加载全部数据导致 JVM 内存溢出OOM传统方案的陷阱简单查询ListUser users userMapper.selectList(queryWrapper);。数据量一大users这个 List 会持有所有数据对象瞬间吃满堆内存。分页查询这是最容易被误解的方案。很多人觉得LIMIT offset, size分页就安全了。但如果你需要全量处理数据比如导出、数据迁移、批量计算分页查询意味着要对数据库进行N次查询N 总数据量 / 每页大小。每次查询都有建立连接、执行SQL、网络传输的开销对数据库是巨大的压力性能极差。调大堆内存-Xmx8g调到-Xmx16g。数据量是无限的内存是有限的。这本质上是把风险后移并且会导致 GC 停顿时间变长影响服务稳定性。流式查询的价值它提供了一种“细水长流”的数据消费模式。数据库服务器端执行查询后并不是一次性发送所有结果而是像打开一个水龙头让客户端你的应用可以一条一条地、或者一小批一小批地拉取数据。在这个过程中在 JVM 内存中同时存在的数据量始终是可控的通常只有几条或一个批次。所以这篇文章要解决的不是“流式查询是什么”的概念问题而是为什么它会成为大数据量查询的救星在MyBatis框架下如何正确地实现它使用它需要注意哪些坑特别是资源泄露除了流式查询还有没有其他备选方案如果你正在开发数据导出、报表生成、ETL任务、大数据量同步等功能或者你的服务经常因为数据查询而内存告警那么这篇文章就是为你写的。2. 基础概念与核心原理在深入代码之前我们必须理解几个关键概念否则很容易误用。2.1 传统查询 vs. 流式查询我们可以用一个快递仓库的比喻来理解传统查询一次性加载你想从仓库数据库取 10000 件商品数据行。仓库管理员数据库驱动的做法是把这 10000 件商品全部打包用一辆巨型卡车一次性运到你家门口应用内存。你的客厅JVM堆必须足够大才能放下所有包裹否则就“爆仓”OOM了。流式查询同样是取 10000 件商品。现在仓库管理员开通了一条传送带数据库游标。他每次只放 1 件或一小箱Fetch Size商品上传送带运到你这里。你拿到一件处理一件或处理一小箱然后告诉传送带“下一件”。你的门口始终只有少量商品客厅永远宽敞。这个“传送带”机制在数据库层面依赖于游标Cursor。2.2 数据库游标Cursor与 Fetch Size游标可以把它想象成数据库结果集的一个“指针”或“书签”。当执行一条查询语句时数据库先在服务端准备好所有符合条件的结果游标初始指向第一条记录之前的位置。Fetch Size获取大小这是客户端JDBC驱动告诉数据库服务器的一个参数“你每次给我发多少条数据”。在传统模式下JDBC驱动可能会设置一个很大的 Fetch Size或者直接让数据库发送所有数据。在流式模式下我们会设置一个较小的、合理的 Fetch Size比如 100、500。这意味着尽管数据库服务端知道所有结果但网络传输和客户端内存中每次只流动一个“批次”的数据。重要关系流式查询的实现本质是通过 JDBC 驱动利用数据库的游标机制并配合合理的Fetch Size来实现的。MyBatis 作为持久层框架是对 JDBC 的封装它提供了更便捷的方式来使用这个特性。2.3 MyBatis 如何支持流式查询MyBatis 提供了两种主要方式来实现流式查询ResultHandler接口这是一种“推送”模式。你实现一个处理器MyBatis 会遍历结果集每获取到一条记录就“推送”给你的处理器回调一次。你可以在回调方法中处理这条数据然后丢弃它。CursorT接口这是一种“拉取”模式。查询返回一个Cursor对象它实现了Iterator接口。你可以像遍历集合一样用hasNext()和next()方法一条一条地“拉取”数据。这种方式更符合编程直觉也是目前更推荐的方式。两者的核心共同点在遍历过程中它们都不会将整个结果集一次性加载到一个List中。数据是“流式”地被消费掉的。理解了这些原理我们就能明白流式查询节省的是JVM 堆内存但它会长时间占用数据库连接和游标资源。这就是它最大的风险点我们会在后面的“坑”里详细讲。3. 环境准备与前置条件在开始编写代码前请确保你的环境满足以下要求。本文以最常见的 Spring Boot MyBatis 组合为例。1. 开发环境与工具JDK8 或以上版本推荐 JDK 11本文示例基于 JDK 8 语法。构建工具Maven 或 Gradle。IDEIntelliJ IDEA, Eclipse, VS Code 等均可。2. 项目依赖Mavenpom.xml你需要引入 Spring Boot、MyBatis 以及数据库驱动。以下是一个基础的依赖配置parent groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version2.7.18/version !-- 请根据实际情况选择稳定版本 -- relativePath/ /parent dependencies !-- Spring Boot Web (如果提供API) -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency !-- MyBatis Spring Boot Starter -- dependency groupIdorg.mybatis.spring.boot/groupId artifactIdmybatis-spring-boot-starter/artifactId version2.3.0/version !-- 请匹配Spring Boot版本 -- /dependency !-- 数据库驱动 (以MySQL为例) -- dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId scoperuntime/scope version8.0.33/version !-- 建议使用8.x版本支持更好的流式特性 -- /dependency !-- Lombok (可选简化实体类) -- dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency !-- 测试 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies3. 数据库准备你需要一个测试表来模拟大数据量。这里创建一个简单的用户表CREATE TABLE large_user ( id bigint(20) NOT NULL AUTO_INCREMENT, name varchar(255) DEFAULT NULL, email varchar(255) DEFAULT NULL, age int(11) DEFAULT NULL, created_at datetime DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 插入测试数据这里可以用存储过程或程序批量插入例如插入100万条。 -- 为了演示我们先理解表结构即可。4. MyBatis 配置application.yml关键的配置在这里它决定了 MyBatis 的默认行为。spring: datasource: url: jdbc:mysql://localhost:3306/your_database?useSSLfalseserverTimezoneUTCuseUnicodetruecharacterEncodingutf8 username: your_username password: your_password driver-class-name: com.mysql.cj.jdbc.Driver # 对于流式查询连接池的配置也很重要 hikari: maximum-pool-size: 10 # 连接池大小根据实际情况调整 connection-timeout: 30000 idle-timeout: 600000 max-lifetime: 1800000 mybatis: configuration: # 非常重要确保下划线转驼峰命名开启方便映射 map-underscore-to-camel-case: true # 默认的 fetchSize对某些驱动有影响。设为负数默认通常使用驱动默认值。 # 对于流式查询我们通常在Mapper方法上通过注解单独设置。 # default-fetch-size: 100 # 指定mapper.xml文件位置 mapper-locations: classpath:mapper/*.xml环境要点总结MySQL 驱动 8.x对游标支持更好。数据库连接池如 HikariCP是生产级应用的标配流式查询会长时间占用连接连接池配置需要合理。MyBatis 的default-fetch-size全局配置需谨慎对于流式查询更推荐在具体方法上通过注解控制。4. 核心流程拆解实现一个安全的流式查询让我们一步步拆解从实体类、Mapper 接口到服务层如何构建一个完整且安全的流式查询流程。4.1 第一步定义实体类对应数据库表large_user。// 文件路径src/main/java/com/example/demo/entity/User.java package com.example.demo.entity; import lombok.Data; import java.time.LocalDateTime; Data public class User { private Long id; private String name; private String email; private Integer age; private LocalDateTime createdAt; }4.2 第二步创建 Mapper 接口与 XML这是实现流式查询的核心。我们将演示两种方式Cursor和ResultHandler。方式一使用CursorT推荐// 文件路径src/main/java/com/example/demo/mapper/UserMapper.java package com.example.demo.mapper; import com.example.demo.entity.User; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Select; import org.apache.ibatis.cursor.Cursor; Mapper public interface UserMapper { /** * 流式查询所有用户使用Cursor * Select注解中通过fetchSize设置获取大小Integer.MIN_VALUE是MySQL驱动的一个特殊值会启用流式模式。 * 也可以设置为一个正数如100。 */ Select(SELECT id, name, email, age, created_at FROM large_user ORDER BY id) CursorUser selectAllUsersStreaming(); }关键点解释Select直接使用注解编写SQL简洁明了。fetchSize Integer.MIN_VALUE这是一个针对 MySQL JDBC 驱动的“魔法值”。设置为此值会告诉驱动我们希望以流式方式获取结果。对于其他数据库如 PostgreSQL可能需要设置一个正整数如fetchSize 100。务必查阅你所使用数据库驱动的官方文档。返回值CursorUser这就是我们的“传送带”手柄。方式二使用ResultHandler传统方式// 在同一个UserMapper接口中增加方法 /** * 流式查询所有用户使用ResultHandler * 注意这个方法返回值为void结果通过handler参数处理。 */ Select(SELECT id, name, email, age, created_at FROM large_user ORDER BY id) void selectAllUsersWithHandler(ResultHandlerUser handler);4.3 第三步编写服务层逻辑服务层负责获取Cursor并安全地遍历它或者使用ResultHandler。服务类使用Cursor// 文件路径src/main/java/com/example/demo/service/UserService.java package com.example.demo.service; import com.example.demo.entity.User; import com.example.demo.mapper.UserMapper; import lombok.extern.slf4j.Slf4j; import org.apache.ibatis.cursor.Cursor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import java.util.ArrayList; import java.util.List; Service Slf4j public class UserService { Autowired private UserMapper userMapper; /** * 使用Cursor进行流式查询并处理数据 * 关键必须在事务内使用Cursor并且确保finally块中关闭Cursor。 */ Transactional(readOnly true) // 只读事务非常重要 public void processUsersWithCursor() { CursorUser cursor null; try { cursor userMapper.selectAllUsersStreaming(); // 获取游标 int count 0; ListUser batch new ArrayList(1000); // 模拟批次处理 for (User user : cursor) { // 遍历Cursor本质是迭代器 // 处理单条数据例如转换、校验、计算 // log.info(Processing user: {}, user.getName()); // 模拟批次处理每1000条执行一次操作如写入文件、发送消息 batch.add(user); if (batch.size() 1000) { processBatch(batch); batch.clear(); } count; } // 处理最后一批 if (!batch.isEmpty()) { processBatch(batch); } log.info(Total users processed: {}, count); } catch (Exception e) { log.error(Error processing users with cursor, e); throw e; // 抛出异常让事务回滚 } finally { // 至关重要显式关闭Cursor释放数据库资源 if (cursor ! null !cursor.isClosed()) { cursor.close(); } } } private void processBatch(ListUser batch) { // 这里实现你的批次处理逻辑例如 // 1. 写入CSV文件 // 2. 批量插入到另一个数据库 // 3. 发送到消息队列 // 4. 进行聚合计算 log.debug(Processing batch of size: {}, batch.size()); // 模拟处理耗时 // try { Thread.sleep(10); } catch (InterruptedException e) { ... } } }代码逻辑拆解与要点Transactional(readOnly true)这是使用Cursor时必须的流式查询需要在同一个数据库连接和事务中完成遍历。如果不在事务中MyBatis 在执行完 Mapper 方法后可能立即关闭连接导致遍历时连接已关闭而报错。readOnlytrue提示这是一个只读事务对性能有一定优化。try-catch-finally块这是资源安全管理的标准模式。确保在任何情况下正常结束或异常都能关闭Cursor。遍历Cursorfor (User user : cursor)语法糖背后就是调用cursor.iterator()。每次next()都会从数据库游标中获取下一条或下一批取决于fetchSize数据。批次处理在循环内直接处理单条数据是可行的但为了提升效率比如减少I/O次数我们通常会将数据累积到一个批次如1000条再进行一次处理processBatch方法。关闭资源cursor.close()会关闭底层的 JDBCResultSet释放数据库游标和相关的服务器端资源。忘记关闭是导致数据库连接泄露的常见原因服务类使用ResultHandler/** * 使用ResultHandler进行流式查询 */ Transactional(readOnly true) public void processUsersWithHandler() { userMapper.selectAllUsersWithHandler(new ResultHandlerUser() { private int count 0; private ListUser batch new ArrayList(1000); Override public void handleResult(ResultContext? extends User resultContext) { User user resultContext.getResultObject(); // 处理单条数据 batch.add(user); if (batch.size() 1000) { processBatch(batch); batch.clear(); } count; // 可以通过resultContext.stop()提前终止 // if (count 10000) { // resultContext.stop(); // } } // 可以重写handleResult的重载方法但通常用这个就够了 }); // ResultHandler的方式由MyBatis自动管理资源通常不需要手动关闭。 log.info(ResultHandler processing finished.); }ResultHandler方式要点推送模式MyBatis 控制循环每得到一条数据就“推送”到你的handleResult方法。资源管理MyBatis 会在方法执行完毕后自动清理ResultSet和Statement通常比Cursor方式更不易泄露资源。灵活性可以通过resultContext.stop()随时停止处理。缺点代码结构更分散不如Cursor的迭代器模式直观且难以在外部控制遍历过程。4.4 第四步创建控制器可选如果你需要通过 HTTP API 触发这个流式处理可以创建一个简单的控制器。// 文件路径src/main/java/com/example/demo/controller/UserController.java package com.example.demo.controller; import com.example.demo.service.UserService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; RestController RequestMapping(/api/users) public class UserController { Autowired private UserService userService; GetMapping(/export) public String exportUsers() { // 注意对于HTTP请求流式查询处理大量数据可能超时。 // 更常见的做法是触发一个异步任务如使用Async或消息队列。 userService.processUsersWithCursor(); return Data export started (streaming mode).; } }HTTP请求警告在同步 HTTP 请求中处理超大数据流很容易导致请求超时。生产环境中这类耗时操作应该改为异步任务立即返回一个任务ID客户端再通过轮询或 WebSocket 获取进度和结果。5. 运行结果与效果验证如何验证我们的流式查询真的在“流式”工作而不是一次性加载我们可以通过观察 JVM 内存使用情况来验证。1. 准备测试数据在large_user表中插入足够多的数据比如 100 万条。可以使用简单的存储过程或编写一个 Java 程序批量插入。2. 编写一个对比测试创建一个传统的查询方法作为对比。// 在UserService中添加 /** * 传统查询一次性加载所有数据危险 */ Transactional(readOnly true) public ListUser getAllUsersTraditional() { // 假设我们有一个返回List的Mapper方法 // Select(SELECT * FROM large_user) // ListUser selectAllUsers(); return userMapper.selectAllUsers(); // 这个方法需要你在Mapper中定义 } public void processUsersTraditional() { ListUser allUsers getAllUsersTraditional(); // 一次性加载到内存 log.info(Loaded {} users into memory., allUsers.size()); // ... 处理数据 }3. 观察内存变化传统方法调用processUsersTraditional()。在启动应用时设置较小的堆内存例如-Xmx256m。当数据量超过内存容量时你会看到控制台抛出java.lang.OutOfMemoryError: Java heap space异常并且在抛出异常前通过 JConsole 或 VisualVM 工具观察会发现堆内存使用率瞬间飙升至接近 100%。流式方法调用processUsersWithCursor()。即使堆内存很小服务也不会 OOM。内存使用曲线会呈现平稳的“锯齿状”——随着批次处理会有小幅的上升和下降由创建对象和GC引起但峰值远低于总堆大小永远不会持续增长到爆掉内存。4. 验证数据库连接流式查询会长时间占用一个数据库连接。你可以通过监控数据库的SHOW PROCESSLIST;命令MySQL来观察。在执行流式查询期间你会看到对应连接的状态一直是Sending data或Writing to net直到遍历结束或游标关闭。成功标志程序能稳定处理远超内存容量的数据量。JVM 内存使用平稳无持续增长。数据处理完毕后数据库连接被正确释放回到连接池。6. 常见问题与排查思路流式查询用起来并不复杂但坑却不少。下面这个表格整理了最常见的问题和解决方法。问题现象可能原因排查方式解决方案Invalid operation for streaming result set或Connection is closed1.未在事务中遍历Cursor。Mapper方法执行完MyBatis就关闭了连接和结果集。2. 在事务方法外获取了Cursor但在遍历时事务已结束。检查调用Cursor遍历的代码是否被Transactional注解包围。检查事务传播行为。确保遍历Cursor的整个逻辑在一个数据库事务内。使用Transactional(readOnlytrue)。数据库连接池连接耗尽 (Timeout waiting for connection)1. 流式查询处理太慢长时间占用连接。2. 忘记关闭Cursor导致连接泄露。3. 连接池maximum-pool-size设置过小。1. 监控连接池活跃连接数。2. 检查代码finally块是否确保cursor.close()。3. 分析处理逻辑是否过慢。1.务必在finally块中关闭Cursor。2. 优化数据处理逻辑加快消费速度。3. 适当增大连接池但根本是解决泄露和慢查询。流式查询速度比普通查询慢很多1.网络往返次数过多如果fetchSize设置过小如1每条数据都产生一次网络I/O。2. 客户端处理逻辑processBatch太慢成了瓶颈。3. 数据库服务器压力大。1. 检查Select注解中的fetchSize值。2. 对处理逻辑进行性能分析。3. 监控数据库服务器负载。1.设置合理的fetchSize如100, 500, 1000。需要在内存和网络I/O间权衡。2. 优化批次处理逻辑考虑异步、多线程处理注意线程安全。内存依然缓慢增长最终OOM1.在遍历过程中将数据累积到了一个大集合中违背了流式初衷。2. 处理逻辑中创建了大量不会被GC的对象如缓存。3. 存在其他内存泄漏。1. 检查processBatch方法确保批次处理完后清空或释放对数据的引用。2. 使用内存分析工具如MAT, JProfiler查看堆转储。1.确保流式消费即处理完的数据立即解除强引用。批次列表在处理后应clear()。2. 检查代码避免在循环内无节制地创建对象。MySQL 报错Commands out of sync; you can‘t run this command now通常是因为在同一连接上前一个流式查询的结果集未处理完就尝试执行新的查询。检查是否在遍历一个Cursor的过程中又用同一个 Mapper/SqlSession 执行了其他数据库操作。1. 确保流式查询在一个独立的事务和方法中完成。2. 避免在遍历循环内调用其他会执行SQL的方法。Cursor.isClosed()返回 false但数据库游标似乎没释放可能是遍历过程发生异常跳过了close()逻辑。检查catch块和finally块的逻辑确保异常时也能执行关闭。使用try-with-resources语法Java 7这是最安全的做法见下文最佳实践。7. 最佳实践与工程建议掌握了基础用法和避坑指南后我们来看看如何将流式查询用得更加优雅和健壮。7.1 使用 Try-With-Resources 自动关闭 Cursor强烈推荐这是 Java 管理资源的标准方式能确保在任何情况下资源都会被关闭。Transactional(readOnly true) public void processUsersWithCursorSafely() { // try-with-resources 语法Cursor实现了AutoCloseable接口 try (CursorUser cursor userMapper.selectAllUsersStreaming()) { int count 0; ListUser batch new ArrayList(1000); for (User user : cursor) { batch.add(user); if (batch.size() 1000) { processBatch(batch); batch.clear(); // 清空列表释放对已处理对象的引用 } count; } processBatch(batch); // 处理最后一批 log.info(Processed {} users safely., count); } catch (Exception e) { log.error(Stream processing failed, e); throw e; // 或进行其他错误处理 } // 无需手动调用 cursor.close()try块结束后会自动调用。 }7.2 为流式查询方法起一个明确的名字在 Mapper 接口中通过方法名清晰表达意图提高代码可读性。// 好名字 CursorUser streamAllUsers(); CursorUser findUsersForExport(Condition condition); // 不够好的名字 CursorUser selectAll(); // 看不出是流式 ListUser selectAllStreaming(); // 返回值是List名字却叫Streaming矛盾7.3 将处理逻辑抽象成策略将“如何处理每一条数据”的逻辑抽象出来使流式查询的框架代码可以复用。public interface StreamingProcessorT { /** * 处理单条数据 */ void processItem(T item); /** * 处理一个批次的数据可选 */ default void processBatch(ListT batch) { for (T item : batch) { processItem(item); } } /** * 所有数据处理完毕后的回调可选 */ default void onFinish() {} } Service public class GenericStreamingService { Autowired private SqlSessionTemplate sqlSessionTemplate; // 用于获取Mapper Transactional(readOnly true) public T void processStreaming(String statementId, StreamingProcessorT processor, int batchSize) { // 通过SqlSession获取Cursor更灵活 try (CursorT cursor sqlSessionTemplate.selectCursor(statementId)) { ListT batch new ArrayList(batchSize); for (T item : cursor) { processor.processItem(item); batch.add(item); if (batch.size() batchSize) { processor.processBatch(batch); batch.clear(); } } if (!batch.isEmpty()) { processor.processBatch(batch); } processor.onFinish(); } catch (Exception e) { // 处理异常 } } } // 使用示例 streamingService.processStreaming( com.example.mapper.UserMapper.streamAllUsers, new StreamingProcessorUser() { Override public void processItem(User user) { // 你的业务逻辑 } }, 1000 );7.4 异步与背压处理对于超大数据量或处理较慢的场景可以考虑异步流式处理并引入背压Backpressure机制防止生产者数据库速度远快于消费者你的处理逻辑导致内存中积压未处理的批次。异步使用Async注解或CompletableFuture将流式处理任务提交到线程池执行避免阻塞主线程或HTTP请求线程。背压在批次处理逻辑中如果队列已满则暂停从Cursor中拉取数据。这通常需要结合有界队列和信号量来实现复杂度较高。一个简单的实践是调整fetchSize和处理批次大小让消费速度能跟上生产速度。7.5 监控与告警在生产环境使用流式查询必须做好监控数据库连接池监控关注活跃连接数、等待连接数。流式查询长时间占用连接是正常现象但数量异常增长可能意味着泄露。应用内存监控观察老年代内存和GC情况确保没有因处理逻辑不当导致的内存缓慢增长。查询超时监控为流式查询设置合理的超时时间可以在数据库连接字符串或MyBatis配置中设置socketTimeout避免一个慢查询永远不释放连接。7.6 明确使用边界什么时候不该用流式查询流式查询不是银弹以下场景请慎重或避免使用数据量很小几千条以内传统方式更简单性能开销更小。需要多次随机访问结果集数据流式查询是单向的只能向前遍历不能回头或跳转。网络环境极差频繁的网络I/O如果fetchSize很小会放大延迟影响。事务隔离级别要求高且处理时间极长长事务会持有锁可能阻塞其他操作。对于 MySQL流式查询默认在REPEATABLE READ隔离级别下会使用一致性读视图可能对性能有影响。8. 总结与后续学习方向通过本文我们深入剖析了 MyBatis 流式查询如何成为应对大数据量查询、防止内存 OOM 的利器。核心要点再回顾一下知其所以然流式查询的核心原理是利用数据库游标和fetchSize实现数据的“按需加载即时消费”从而将内存占用从O(N)降低到O(1)或O(BatchSize)。正确使用在 MyBatis 中优先使用CursorT接口并务必将其置于Transactional事务内使用try-with-resources语法确保资源关闭。规避风险最大的风险是数据库连接泄露和长事务。务必关闭Cursor合理设置连接池和超时并优化数据处理速度。最佳实践抽象处理逻辑、合理设置批次大小、做好监控、明确适用场景。流式查询解决了“内存放不下”的问题但它把压力转移到了“数据库连接占用时间”和“网络I/O次数”上。这是一个典型的权衡。后续你可以深入探索的方向数据库方言差异不同数据库PostgreSQL, Oracle, SQL Server对游标和流式查询的支持方式、fetchSize的语义可能不同需要查阅对应驱动的文档。与 Spring Data JPA 的结合如果你使用 JPA可以研究如何通过ScrollableResults或 Hibernate 的StatelessSession实现类似功能。响应式编程集成在 Spring WebFlux 项目中可以将Cursor转换为Flux实现真正的响应式数据流处理。更复杂的数据管道结合 Apache Spark、Flink 或简单的 Spring Batch将流式查询作为数据源构建更强大的批处理或流处理任务。处理海量数据是现代后端开发的必修课。流式查询是工具箱中一件关键且实用的武器。理解其原理掌握其正确用法警惕其陷阱你就能在“数据洪流”面前从容地打开那道安全阀让数据平稳、高效地流过你的系统。
MyBatis流式查询实战：大数据量查询防OOM的核心原理与安全实现
你有没有遇到过这样的场景一个看似简单的数据导出功能在测试环境跑得好好的一到生产环境就突然内存飙升直接 OOMOutOfMemoryError把服务干趴下你查了半天日志发现罪魁祸首只是一条平平无奇的SELECT * FROM large_table。这不是段子而是很多后端开发者踩过的真实大坑。当数据量从几千条变成几百万条时传统的 JDBC 查询方式会瞬间将海量数据全部加载到 JVM 内存中内存被“挤爆”几乎是必然结果。很多人第一反应是调大 JVM 堆内存但这只是饮鸩止渴数据再大一点呢今天要聊的MyBatis 流式查询就是专门用来“拆弹”的。它不是什么高深的新技术却是处理大数据量查询时避免内存溢出的“标准答案”。很多人知道这个概念但在实际项目中要么不敢用要么用错了反而引入了连接泄露的新问题。本文将彻底讲清楚为什么一行普通的查询代码能“挤爆”内存MyBatis 流式查询的原理是什么如何正确、安全地使用它以及在什么场景下你其实根本不需要它。1. 这篇文章真正要解决的问题我们首先得达成一个共识技术方案的选择永远是对“成本”的权衡。这里的成本包括内存成本、CPU成本、网络I/O成本以及最重要的——开发和维护的复杂度成本。核心问题当你的 Java 应用需要从数据库查询并处理大量数据比如百万级以上时如何避免一次性加载全部数据导致 JVM 内存溢出OOM传统方案的陷阱简单查询ListUser users userMapper.selectList(queryWrapper);。数据量一大users这个 List 会持有所有数据对象瞬间吃满堆内存。分页查询这是最容易被误解的方案。很多人觉得LIMIT offset, size分页就安全了。但如果你需要全量处理数据比如导出、数据迁移、批量计算分页查询意味着要对数据库进行N次查询N 总数据量 / 每页大小。每次查询都有建立连接、执行SQL、网络传输的开销对数据库是巨大的压力性能极差。调大堆内存-Xmx8g调到-Xmx16g。数据量是无限的内存是有限的。这本质上是把风险后移并且会导致 GC 停顿时间变长影响服务稳定性。流式查询的价值它提供了一种“细水长流”的数据消费模式。数据库服务器端执行查询后并不是一次性发送所有结果而是像打开一个水龙头让客户端你的应用可以一条一条地、或者一小批一小批地拉取数据。在这个过程中在 JVM 内存中同时存在的数据量始终是可控的通常只有几条或一个批次。所以这篇文章要解决的不是“流式查询是什么”的概念问题而是为什么它会成为大数据量查询的救星在MyBatis框架下如何正确地实现它使用它需要注意哪些坑特别是资源泄露除了流式查询还有没有其他备选方案如果你正在开发数据导出、报表生成、ETL任务、大数据量同步等功能或者你的服务经常因为数据查询而内存告警那么这篇文章就是为你写的。2. 基础概念与核心原理在深入代码之前我们必须理解几个关键概念否则很容易误用。2.1 传统查询 vs. 流式查询我们可以用一个快递仓库的比喻来理解传统查询一次性加载你想从仓库数据库取 10000 件商品数据行。仓库管理员数据库驱动的做法是把这 10000 件商品全部打包用一辆巨型卡车一次性运到你家门口应用内存。你的客厅JVM堆必须足够大才能放下所有包裹否则就“爆仓”OOM了。流式查询同样是取 10000 件商品。现在仓库管理员开通了一条传送带数据库游标。他每次只放 1 件或一小箱Fetch Size商品上传送带运到你这里。你拿到一件处理一件或处理一小箱然后告诉传送带“下一件”。你的门口始终只有少量商品客厅永远宽敞。这个“传送带”机制在数据库层面依赖于游标Cursor。2.2 数据库游标Cursor与 Fetch Size游标可以把它想象成数据库结果集的一个“指针”或“书签”。当执行一条查询语句时数据库先在服务端准备好所有符合条件的结果游标初始指向第一条记录之前的位置。Fetch Size获取大小这是客户端JDBC驱动告诉数据库服务器的一个参数“你每次给我发多少条数据”。在传统模式下JDBC驱动可能会设置一个很大的 Fetch Size或者直接让数据库发送所有数据。在流式模式下我们会设置一个较小的、合理的 Fetch Size比如 100、500。这意味着尽管数据库服务端知道所有结果但网络传输和客户端内存中每次只流动一个“批次”的数据。重要关系流式查询的实现本质是通过 JDBC 驱动利用数据库的游标机制并配合合理的Fetch Size来实现的。MyBatis 作为持久层框架是对 JDBC 的封装它提供了更便捷的方式来使用这个特性。2.3 MyBatis 如何支持流式查询MyBatis 提供了两种主要方式来实现流式查询ResultHandler接口这是一种“推送”模式。你实现一个处理器MyBatis 会遍历结果集每获取到一条记录就“推送”给你的处理器回调一次。你可以在回调方法中处理这条数据然后丢弃它。CursorT接口这是一种“拉取”模式。查询返回一个Cursor对象它实现了Iterator接口。你可以像遍历集合一样用hasNext()和next()方法一条一条地“拉取”数据。这种方式更符合编程直觉也是目前更推荐的方式。两者的核心共同点在遍历过程中它们都不会将整个结果集一次性加载到一个List中。数据是“流式”地被消费掉的。理解了这些原理我们就能明白流式查询节省的是JVM 堆内存但它会长时间占用数据库连接和游标资源。这就是它最大的风险点我们会在后面的“坑”里详细讲。3. 环境准备与前置条件在开始编写代码前请确保你的环境满足以下要求。本文以最常见的 Spring Boot MyBatis 组合为例。1. 开发环境与工具JDK8 或以上版本推荐 JDK 11本文示例基于 JDK 8 语法。构建工具Maven 或 Gradle。IDEIntelliJ IDEA, Eclipse, VS Code 等均可。2. 项目依赖Mavenpom.xml你需要引入 Spring Boot、MyBatis 以及数据库驱动。以下是一个基础的依赖配置parent groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-parent/artifactId version2.7.18/version !-- 请根据实际情况选择稳定版本 -- relativePath/ /parent dependencies !-- Spring Boot Web (如果提供API) -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency !-- MyBatis Spring Boot Starter -- dependency groupIdorg.mybatis.spring.boot/groupId artifactIdmybatis-spring-boot-starter/artifactId version2.3.0/version !-- 请匹配Spring Boot版本 -- /dependency !-- 数据库驱动 (以MySQL为例) -- dependency groupIdmysql/groupId artifactIdmysql-connector-java/artifactId scoperuntime/scope version8.0.33/version !-- 建议使用8.x版本支持更好的流式特性 -- /dependency !-- Lombok (可选简化实体类) -- dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency !-- 测试 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-test/artifactId scopetest/scope /dependency /dependencies3. 数据库准备你需要一个测试表来模拟大数据量。这里创建一个简单的用户表CREATE TABLE large_user ( id bigint(20) NOT NULL AUTO_INCREMENT, name varchar(255) DEFAULT NULL, email varchar(255) DEFAULT NULL, age int(11) DEFAULT NULL, created_at datetime DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; -- 插入测试数据这里可以用存储过程或程序批量插入例如插入100万条。 -- 为了演示我们先理解表结构即可。4. MyBatis 配置application.yml关键的配置在这里它决定了 MyBatis 的默认行为。spring: datasource: url: jdbc:mysql://localhost:3306/your_database?useSSLfalseserverTimezoneUTCuseUnicodetruecharacterEncodingutf8 username: your_username password: your_password driver-class-name: com.mysql.cj.jdbc.Driver # 对于流式查询连接池的配置也很重要 hikari: maximum-pool-size: 10 # 连接池大小根据实际情况调整 connection-timeout: 30000 idle-timeout: 600000 max-lifetime: 1800000 mybatis: configuration: # 非常重要确保下划线转驼峰命名开启方便映射 map-underscore-to-camel-case: true # 默认的 fetchSize对某些驱动有影响。设为负数默认通常使用驱动默认值。 # 对于流式查询我们通常在Mapper方法上通过注解单独设置。 # default-fetch-size: 100 # 指定mapper.xml文件位置 mapper-locations: classpath:mapper/*.xml环境要点总结MySQL 驱动 8.x对游标支持更好。数据库连接池如 HikariCP是生产级应用的标配流式查询会长时间占用连接连接池配置需要合理。MyBatis 的default-fetch-size全局配置需谨慎对于流式查询更推荐在具体方法上通过注解控制。4. 核心流程拆解实现一个安全的流式查询让我们一步步拆解从实体类、Mapper 接口到服务层如何构建一个完整且安全的流式查询流程。4.1 第一步定义实体类对应数据库表large_user。// 文件路径src/main/java/com/example/demo/entity/User.java package com.example.demo.entity; import lombok.Data; import java.time.LocalDateTime; Data public class User { private Long id; private String name; private String email; private Integer age; private LocalDateTime createdAt; }4.2 第二步创建 Mapper 接口与 XML这是实现流式查询的核心。我们将演示两种方式Cursor和ResultHandler。方式一使用CursorT推荐// 文件路径src/main/java/com/example/demo/mapper/UserMapper.java package com.example.demo.mapper; import com.example.demo.entity.User; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Select; import org.apache.ibatis.cursor.Cursor; Mapper public interface UserMapper { /** * 流式查询所有用户使用Cursor * Select注解中通过fetchSize设置获取大小Integer.MIN_VALUE是MySQL驱动的一个特殊值会启用流式模式。 * 也可以设置为一个正数如100。 */ Select(SELECT id, name, email, age, created_at FROM large_user ORDER BY id) CursorUser selectAllUsersStreaming(); }关键点解释Select直接使用注解编写SQL简洁明了。fetchSize Integer.MIN_VALUE这是一个针对 MySQL JDBC 驱动的“魔法值”。设置为此值会告诉驱动我们希望以流式方式获取结果。对于其他数据库如 PostgreSQL可能需要设置一个正整数如fetchSize 100。务必查阅你所使用数据库驱动的官方文档。返回值CursorUser这就是我们的“传送带”手柄。方式二使用ResultHandler传统方式// 在同一个UserMapper接口中增加方法 /** * 流式查询所有用户使用ResultHandler * 注意这个方法返回值为void结果通过handler参数处理。 */ Select(SELECT id, name, email, age, created_at FROM large_user ORDER BY id) void selectAllUsersWithHandler(ResultHandlerUser handler);4.3 第三步编写服务层逻辑服务层负责获取Cursor并安全地遍历它或者使用ResultHandler。服务类使用Cursor// 文件路径src/main/java/com/example/demo/service/UserService.java package com.example.demo.service; import com.example.demo.entity.User; import com.example.demo.mapper.UserMapper; import lombok.extern.slf4j.Slf4j; import org.apache.ibatis.cursor.Cursor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import java.util.ArrayList; import java.util.List; Service Slf4j public class UserService { Autowired private UserMapper userMapper; /** * 使用Cursor进行流式查询并处理数据 * 关键必须在事务内使用Cursor并且确保finally块中关闭Cursor。 */ Transactional(readOnly true) // 只读事务非常重要 public void processUsersWithCursor() { CursorUser cursor null; try { cursor userMapper.selectAllUsersStreaming(); // 获取游标 int count 0; ListUser batch new ArrayList(1000); // 模拟批次处理 for (User user : cursor) { // 遍历Cursor本质是迭代器 // 处理单条数据例如转换、校验、计算 // log.info(Processing user: {}, user.getName()); // 模拟批次处理每1000条执行一次操作如写入文件、发送消息 batch.add(user); if (batch.size() 1000) { processBatch(batch); batch.clear(); } count; } // 处理最后一批 if (!batch.isEmpty()) { processBatch(batch); } log.info(Total users processed: {}, count); } catch (Exception e) { log.error(Error processing users with cursor, e); throw e; // 抛出异常让事务回滚 } finally { // 至关重要显式关闭Cursor释放数据库资源 if (cursor ! null !cursor.isClosed()) { cursor.close(); } } } private void processBatch(ListUser batch) { // 这里实现你的批次处理逻辑例如 // 1. 写入CSV文件 // 2. 批量插入到另一个数据库 // 3. 发送到消息队列 // 4. 进行聚合计算 log.debug(Processing batch of size: {}, batch.size()); // 模拟处理耗时 // try { Thread.sleep(10); } catch (InterruptedException e) { ... } } }代码逻辑拆解与要点Transactional(readOnly true)这是使用Cursor时必须的流式查询需要在同一个数据库连接和事务中完成遍历。如果不在事务中MyBatis 在执行完 Mapper 方法后可能立即关闭连接导致遍历时连接已关闭而报错。readOnlytrue提示这是一个只读事务对性能有一定优化。try-catch-finally块这是资源安全管理的标准模式。确保在任何情况下正常结束或异常都能关闭Cursor。遍历Cursorfor (User user : cursor)语法糖背后就是调用cursor.iterator()。每次next()都会从数据库游标中获取下一条或下一批取决于fetchSize数据。批次处理在循环内直接处理单条数据是可行的但为了提升效率比如减少I/O次数我们通常会将数据累积到一个批次如1000条再进行一次处理processBatch方法。关闭资源cursor.close()会关闭底层的 JDBCResultSet释放数据库游标和相关的服务器端资源。忘记关闭是导致数据库连接泄露的常见原因服务类使用ResultHandler/** * 使用ResultHandler进行流式查询 */ Transactional(readOnly true) public void processUsersWithHandler() { userMapper.selectAllUsersWithHandler(new ResultHandlerUser() { private int count 0; private ListUser batch new ArrayList(1000); Override public void handleResult(ResultContext? extends User resultContext) { User user resultContext.getResultObject(); // 处理单条数据 batch.add(user); if (batch.size() 1000) { processBatch(batch); batch.clear(); } count; // 可以通过resultContext.stop()提前终止 // if (count 10000) { // resultContext.stop(); // } } // 可以重写handleResult的重载方法但通常用这个就够了 }); // ResultHandler的方式由MyBatis自动管理资源通常不需要手动关闭。 log.info(ResultHandler processing finished.); }ResultHandler方式要点推送模式MyBatis 控制循环每得到一条数据就“推送”到你的handleResult方法。资源管理MyBatis 会在方法执行完毕后自动清理ResultSet和Statement通常比Cursor方式更不易泄露资源。灵活性可以通过resultContext.stop()随时停止处理。缺点代码结构更分散不如Cursor的迭代器模式直观且难以在外部控制遍历过程。4.4 第四步创建控制器可选如果你需要通过 HTTP API 触发这个流式处理可以创建一个简单的控制器。// 文件路径src/main/java/com/example/demo/controller/UserController.java package com.example.demo.controller; import com.example.demo.service.UserService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; RestController RequestMapping(/api/users) public class UserController { Autowired private UserService userService; GetMapping(/export) public String exportUsers() { // 注意对于HTTP请求流式查询处理大量数据可能超时。 // 更常见的做法是触发一个异步任务如使用Async或消息队列。 userService.processUsersWithCursor(); return Data export started (streaming mode).; } }HTTP请求警告在同步 HTTP 请求中处理超大数据流很容易导致请求超时。生产环境中这类耗时操作应该改为异步任务立即返回一个任务ID客户端再通过轮询或 WebSocket 获取进度和结果。5. 运行结果与效果验证如何验证我们的流式查询真的在“流式”工作而不是一次性加载我们可以通过观察 JVM 内存使用情况来验证。1. 准备测试数据在large_user表中插入足够多的数据比如 100 万条。可以使用简单的存储过程或编写一个 Java 程序批量插入。2. 编写一个对比测试创建一个传统的查询方法作为对比。// 在UserService中添加 /** * 传统查询一次性加载所有数据危险 */ Transactional(readOnly true) public ListUser getAllUsersTraditional() { // 假设我们有一个返回List的Mapper方法 // Select(SELECT * FROM large_user) // ListUser selectAllUsers(); return userMapper.selectAllUsers(); // 这个方法需要你在Mapper中定义 } public void processUsersTraditional() { ListUser allUsers getAllUsersTraditional(); // 一次性加载到内存 log.info(Loaded {} users into memory., allUsers.size()); // ... 处理数据 }3. 观察内存变化传统方法调用processUsersTraditional()。在启动应用时设置较小的堆内存例如-Xmx256m。当数据量超过内存容量时你会看到控制台抛出java.lang.OutOfMemoryError: Java heap space异常并且在抛出异常前通过 JConsole 或 VisualVM 工具观察会发现堆内存使用率瞬间飙升至接近 100%。流式方法调用processUsersWithCursor()。即使堆内存很小服务也不会 OOM。内存使用曲线会呈现平稳的“锯齿状”——随着批次处理会有小幅的上升和下降由创建对象和GC引起但峰值远低于总堆大小永远不会持续增长到爆掉内存。4. 验证数据库连接流式查询会长时间占用一个数据库连接。你可以通过监控数据库的SHOW PROCESSLIST;命令MySQL来观察。在执行流式查询期间你会看到对应连接的状态一直是Sending data或Writing to net直到遍历结束或游标关闭。成功标志程序能稳定处理远超内存容量的数据量。JVM 内存使用平稳无持续增长。数据处理完毕后数据库连接被正确释放回到连接池。6. 常见问题与排查思路流式查询用起来并不复杂但坑却不少。下面这个表格整理了最常见的问题和解决方法。问题现象可能原因排查方式解决方案Invalid operation for streaming result set或Connection is closed1.未在事务中遍历Cursor。Mapper方法执行完MyBatis就关闭了连接和结果集。2. 在事务方法外获取了Cursor但在遍历时事务已结束。检查调用Cursor遍历的代码是否被Transactional注解包围。检查事务传播行为。确保遍历Cursor的整个逻辑在一个数据库事务内。使用Transactional(readOnlytrue)。数据库连接池连接耗尽 (Timeout waiting for connection)1. 流式查询处理太慢长时间占用连接。2. 忘记关闭Cursor导致连接泄露。3. 连接池maximum-pool-size设置过小。1. 监控连接池活跃连接数。2. 检查代码finally块是否确保cursor.close()。3. 分析处理逻辑是否过慢。1.务必在finally块中关闭Cursor。2. 优化数据处理逻辑加快消费速度。3. 适当增大连接池但根本是解决泄露和慢查询。流式查询速度比普通查询慢很多1.网络往返次数过多如果fetchSize设置过小如1每条数据都产生一次网络I/O。2. 客户端处理逻辑processBatch太慢成了瓶颈。3. 数据库服务器压力大。1. 检查Select注解中的fetchSize值。2. 对处理逻辑进行性能分析。3. 监控数据库服务器负载。1.设置合理的fetchSize如100, 500, 1000。需要在内存和网络I/O间权衡。2. 优化批次处理逻辑考虑异步、多线程处理注意线程安全。内存依然缓慢增长最终OOM1.在遍历过程中将数据累积到了一个大集合中违背了流式初衷。2. 处理逻辑中创建了大量不会被GC的对象如缓存。3. 存在其他内存泄漏。1. 检查processBatch方法确保批次处理完后清空或释放对数据的引用。2. 使用内存分析工具如MAT, JProfiler查看堆转储。1.确保流式消费即处理完的数据立即解除强引用。批次列表在处理后应clear()。2. 检查代码避免在循环内无节制地创建对象。MySQL 报错Commands out of sync; you can‘t run this command now通常是因为在同一连接上前一个流式查询的结果集未处理完就尝试执行新的查询。检查是否在遍历一个Cursor的过程中又用同一个 Mapper/SqlSession 执行了其他数据库操作。1. 确保流式查询在一个独立的事务和方法中完成。2. 避免在遍历循环内调用其他会执行SQL的方法。Cursor.isClosed()返回 false但数据库游标似乎没释放可能是遍历过程发生异常跳过了close()逻辑。检查catch块和finally块的逻辑确保异常时也能执行关闭。使用try-with-resources语法Java 7这是最安全的做法见下文最佳实践。7. 最佳实践与工程建议掌握了基础用法和避坑指南后我们来看看如何将流式查询用得更加优雅和健壮。7.1 使用 Try-With-Resources 自动关闭 Cursor强烈推荐这是 Java 管理资源的标准方式能确保在任何情况下资源都会被关闭。Transactional(readOnly true) public void processUsersWithCursorSafely() { // try-with-resources 语法Cursor实现了AutoCloseable接口 try (CursorUser cursor userMapper.selectAllUsersStreaming()) { int count 0; ListUser batch new ArrayList(1000); for (User user : cursor) { batch.add(user); if (batch.size() 1000) { processBatch(batch); batch.clear(); // 清空列表释放对已处理对象的引用 } count; } processBatch(batch); // 处理最后一批 log.info(Processed {} users safely., count); } catch (Exception e) { log.error(Stream processing failed, e); throw e; // 或进行其他错误处理 } // 无需手动调用 cursor.close()try块结束后会自动调用。 }7.2 为流式查询方法起一个明确的名字在 Mapper 接口中通过方法名清晰表达意图提高代码可读性。// 好名字 CursorUser streamAllUsers(); CursorUser findUsersForExport(Condition condition); // 不够好的名字 CursorUser selectAll(); // 看不出是流式 ListUser selectAllStreaming(); // 返回值是List名字却叫Streaming矛盾7.3 将处理逻辑抽象成策略将“如何处理每一条数据”的逻辑抽象出来使流式查询的框架代码可以复用。public interface StreamingProcessorT { /** * 处理单条数据 */ void processItem(T item); /** * 处理一个批次的数据可选 */ default void processBatch(ListT batch) { for (T item : batch) { processItem(item); } } /** * 所有数据处理完毕后的回调可选 */ default void onFinish() {} } Service public class GenericStreamingService { Autowired private SqlSessionTemplate sqlSessionTemplate; // 用于获取Mapper Transactional(readOnly true) public T void processStreaming(String statementId, StreamingProcessorT processor, int batchSize) { // 通过SqlSession获取Cursor更灵活 try (CursorT cursor sqlSessionTemplate.selectCursor(statementId)) { ListT batch new ArrayList(batchSize); for (T item : cursor) { processor.processItem(item); batch.add(item); if (batch.size() batchSize) { processor.processBatch(batch); batch.clear(); } } if (!batch.isEmpty()) { processor.processBatch(batch); } processor.onFinish(); } catch (Exception e) { // 处理异常 } } } // 使用示例 streamingService.processStreaming( com.example.mapper.UserMapper.streamAllUsers, new StreamingProcessorUser() { Override public void processItem(User user) { // 你的业务逻辑 } }, 1000 );7.4 异步与背压处理对于超大数据量或处理较慢的场景可以考虑异步流式处理并引入背压Backpressure机制防止生产者数据库速度远快于消费者你的处理逻辑导致内存中积压未处理的批次。异步使用Async注解或CompletableFuture将流式处理任务提交到线程池执行避免阻塞主线程或HTTP请求线程。背压在批次处理逻辑中如果队列已满则暂停从Cursor中拉取数据。这通常需要结合有界队列和信号量来实现复杂度较高。一个简单的实践是调整fetchSize和处理批次大小让消费速度能跟上生产速度。7.5 监控与告警在生产环境使用流式查询必须做好监控数据库连接池监控关注活跃连接数、等待连接数。流式查询长时间占用连接是正常现象但数量异常增长可能意味着泄露。应用内存监控观察老年代内存和GC情况确保没有因处理逻辑不当导致的内存缓慢增长。查询超时监控为流式查询设置合理的超时时间可以在数据库连接字符串或MyBatis配置中设置socketTimeout避免一个慢查询永远不释放连接。7.6 明确使用边界什么时候不该用流式查询流式查询不是银弹以下场景请慎重或避免使用数据量很小几千条以内传统方式更简单性能开销更小。需要多次随机访问结果集数据流式查询是单向的只能向前遍历不能回头或跳转。网络环境极差频繁的网络I/O如果fetchSize很小会放大延迟影响。事务隔离级别要求高且处理时间极长长事务会持有锁可能阻塞其他操作。对于 MySQL流式查询默认在REPEATABLE READ隔离级别下会使用一致性读视图可能对性能有影响。8. 总结与后续学习方向通过本文我们深入剖析了 MyBatis 流式查询如何成为应对大数据量查询、防止内存 OOM 的利器。核心要点再回顾一下知其所以然流式查询的核心原理是利用数据库游标和fetchSize实现数据的“按需加载即时消费”从而将内存占用从O(N)降低到O(1)或O(BatchSize)。正确使用在 MyBatis 中优先使用CursorT接口并务必将其置于Transactional事务内使用try-with-resources语法确保资源关闭。规避风险最大的风险是数据库连接泄露和长事务。务必关闭Cursor合理设置连接池和超时并优化数据处理速度。最佳实践抽象处理逻辑、合理设置批次大小、做好监控、明确适用场景。流式查询解决了“内存放不下”的问题但它把压力转移到了“数据库连接占用时间”和“网络I/O次数”上。这是一个典型的权衡。后续你可以深入探索的方向数据库方言差异不同数据库PostgreSQL, Oracle, SQL Server对游标和流式查询的支持方式、fetchSize的语义可能不同需要查阅对应驱动的文档。与 Spring Data JPA 的结合如果你使用 JPA可以研究如何通过ScrollableResults或 Hibernate 的StatelessSession实现类似功能。响应式编程集成在 Spring WebFlux 项目中可以将Cursor转换为Flux实现真正的响应式数据流处理。更复杂的数据管道结合 Apache Spark、Flink 或简单的 Spring Batch将流式查询作为数据源构建更强大的批处理或流处理任务。处理海量数据是现代后端开发的必修课。流式查询是工具箱中一件关键且实用的武器。理解其原理掌握其正确用法警惕其陷阱你就能在“数据洪流”面前从容地打开那道安全阀让数据平稳、高效地流过你的系统。
