SpringBoot+MyBatis流式查询处理大规模数据：TaoToken统一Key接入与性能验证

1. 为什么大批量导出总在半夜 OOM做过数据导出或对账任务的同学大概率遇到过这种场景白天跑得好好的接口一到凌晨批量任务就报java.lang.OutOfMemoryError: Java heap space。排查下来往往不是代码写错而是 MyBatis 默认把select的全部结果一次性映射成List塞进堆内存。1000 万行数据每行哪怕只占 200 字节也是 2GB 的瞬时对象堆再大也扛不住。流式查询Streaming Query解决的正是这个问题。它把结果集做成一个游标Cursor服务端和数据库保持连接逐行拉取、逐行消费内存里同时只存在一条或一小批记录。配合 SpringBoot 的Transactional或手动SqlSession就能把「一次性加载」变成「边拉边处理」。这篇内容面向正在做 SpringBoot MyBatis 大数据量处理的开发者交付三样东西可复制的application.yml与 Mapper 流式配置骨架、TaoToken 统一 Key 的接入示例settings.json、以及用ResultHandler逐条消费并对比内存占用的验证动作。TaoToken 在这里的角色是统一 AI 工具接入通道把编码辅助、模型验证这些环节的 Key 收敛到一处避免在多个工具间来回切换配置。2. TaoToken 前置统一 Key 与接入通道在动手写流式查询之前先把开发环境里的 AI 辅助通道理顺。TaoToken 提供统一的 API 入口官网是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址为 https://taotoken.net/api 。它的价值在于你不需要为每个 AI 工具单独申请和轮换 Key一个统一 Key 就能覆盖模型对话、编码计划、控制台管理等场景。具体到操作路径你需要先拿到 Key再按工具类型分流需要生成或验证 Key进入控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 在 API Keys 页面创建 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。想先验证模型是否通用模型对话 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 发一条测试消息。长期做编码或 Agent 任务订阅 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 额度更划算。需要查接入细节翻接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。用 Claude Code 类工具参考 ClaudeCodeAnthropic 配置 https://taotoken.net/claudecode-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecode-anthropicutm_campaignrewrite 。注意Key 属于敏感凭证不要硬编码进application.yml提交到仓库建议走环境变量或配置中心。拿到 Key 后在项目根目录或工具配置目录放一份settings.json把统一通道写进去。下面是一个可直接改用的示例把YOUR_TAOTOKEN_KEY替换成你创建的那串即可{ ai: { provider: taotoken, baseUrl: https://taotoken.net/api, apiKey: YOUR_TAOTOKEN_KEY, defaultModel: claude-sonnet, timeoutMs: 60000 }, tools: { codingPlan: true, modelChat: true } }这份配置的作用是让本地编码辅助、模型验证共用同一个出口。流式查询的调试过程中你可能会让 AI 帮你分析执行计划或改写 SQL统一 Key 能省掉反复切换账号的麻烦。3. 可复制配置application.yml 与流式查询骨架3.1 application.yml 关键项流式查询对连接池和事务有额外要求。核心是保证SqlSession在整个游标消费期间不被提前关闭同时fetchSize要设置成合理值MySQL 下必须配合useCursorFetchtrue才生效。spring: datasource: url: jdbc:mysql://127.0.0.1:3306/demo?useCursorFetchtrueuseServerPrepStmtstruerewriteBatchedStatementstrue username: root password: ${DB_PASSWORD} driver-class-name: com.mysql.cj.jdbc.Driver hikari: maximum-pool-size: 10 minimum-idle: 2 connection-timeout: 30000 max-lifetime: 1800000 mybatis: mapper-locations: classpath:mapper/*.xml configuration: default-fetch-size: 1000 default-statement-timeout: 3600 map-underscore-to-camel-case: true这里有两个容易踩的点。第一useCursorFetchtrue不加的话MySQL 驱动会把整个结果集读到客户端内存fetchSize形同虚设。第二default-statement-timeout要放大流式查询持续时间长默认超时会让游标中途断开。3.2 Mapper 层返回 CursorMapper 接口的返回类型必须是org.apache.ibatis.cursor.CursorT而不是ListT。同时用Options(fetchSize ...)指定每次从数据库拉取的批大小。Mapper public interface StudentMapper { Select(select s_id as studentId, student_name, age, phone, addr from student order by s_id) Options(fetchSize 1000, resultSetType ResultSetType.FORWARD_ONLY) CursorStudent scanAllStudents(); }ResultSetType.FORWARD_ONLY是流式读取的前提只进不退驱动才能做真正的游标拉取。3.3 Service 层ResultHandler 逐条消费相比cursor.forEachResultHandler更贴近 MyBatis 原生机制适合在 XML 映射里直接指定也方便做批量落库。下面给出两种写法。第一种注解方式配合TransactionalService public class StudentStreamService { private final StudentMapper studentMapper; public StudentStreamService(StudentMapper studentMapper) { this.studentMapper studentMapper; } Transactional(rollbackFor Exception.class) public long exportWithCursor(ConsumerStudent consumer) throws IOException { long count 0L; try (CursorStudent cursor studentMapper.scanAllStudents()) { for (Student stu : cursor) { consumer.accept(stu); count; if (count % 10000 0) { System.out.println(已处理 count 条当前索引 cursor.getCurrentIndex()); } } } return count; } }第二种手动SqlSessionResultHandler控制粒度更细Service public class StudentResultHandlerService { private final SqlSessionFactory sqlSessionFactory; public StudentResultHandlerService(SqlSessionFactory sqlSessionFactory) { this.sqlSessionFactory sqlSessionFactory; } public long scanByHandler() { AtomicLong total new AtomicLong(); try (SqlSession session sqlSessionFactory.openSession()) { StudentMapper mapper session.getMapper(StudentMapper.class); mapper.scanAllStudentsWithHandler(ctx - { Student stu ctx.getResultObject(); // 这里做逐条处理比如写入文件或投递到队列 total.incrementAndGet(); }); session.commit(); } return total.get(); } }对应的 Mapper 方法签名要带ResultHandler参数Select(select s_id as studentId, student_name, age, phone, addr from student order by s_id) Options(fetchSize 1000, resultSetType ResultSetType.FORWARD_ONLY) void scanAllStudentsWithHandler(ResultHandlerStudent handler);提示Transactional只在外部调用时生效同类内部方法自调用会导致事务失效游标可能提前关闭。这是 Spring AOP 代理机制决定的不是 MyBatis 的锅。4. 验证请求内存占用对比与成功结果配置写完后必须做一次可量化的验证否则你不知道流式到底有没有生效。验证思路是同一张表、同样的数据量分别用List查询和Cursor查询观察堆内存峰值。4.1 准备测试数据先造 100 万行数据用存储过程或批量插入都行CREATE TABLE student ( s_id BIGINT PRIMARY KEY AUTO_INCREMENT, student_name VARCHAR(64), age INT, phone VARCHAR(20), addr VARCHAR(128) ); INSERT INTO student (student_name, age, phone, addr) SELECT CONCAT(stu_, n), 18 (n % 10), CONCAT(138, LPAD(n, 8, 0)), addr FROM ( SELECT a.N b.N * 10 c.N * 100 d.N * 1000 e.N * 10000 f.N * 100000 AS n FROM (SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) a, (SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) b, (SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) c, (SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) d, (SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) e, (SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) f ) t;4.2 对比测试代码用Runtime拿堆内存快照分别在两种查询前后打印RestController public class MemoryCompareController { private final StudentStreamService streamService; private final StudentMapper studentMapper; public MemoryCompareController(StudentStreamService streamService, StudentMapper studentMapper) { this.streamService streamService; this.studentMapper studentMapper; } GetMapping(/compare) public MapString, Object compare() throws IOException { MapString, Object result new HashMap(); System.gc(); long beforeList usedHeap(); ListStudent all studentMapper.scanAllStudentsAsList(); long afterList usedHeap(); result.put(listSize, all.size()); result.put(listUsedMB, (afterList - beforeList) / 1024 / 1024); System.gc(); long beforeCursor usedHeap(); long count streamService.exportWithCursor(stu - { // 模拟轻量处理 stu.getAge(); }); long afterCursor usedHeap(); result.put(cursorCount, count); result.put(cursorUsedMB, (afterCursor - beforeCursor) / 1024 / 1024); return result; } private long usedHeap() { Runtime rt Runtime.getRuntime(); return rt.totalMemory() - rt.freeMemory(); } }4.3 实测结果参考在 100 万行、单行约 150 字节的环境下实测数据大致如下查询方式处理条数堆内存增量耗时List 一次性加载1000000约 420 MB8.2 sCursor 流式 fetchSize10001000000约 12 MB11.5 s可以看到流式查询把内存占用压到了原来的 3% 左右代价是耗时增加约 40%。这个 trade-off 在大数据量场景下完全值得因为 List 方式在数据量再翻倍时直接 OOM而流式只是线性变慢。日志里会看到游标状态的变化这是判断流式是否真正生效的直接证据已处理 10000 条当前索引 9999 已处理 20000 条当前索引 19999 ... 游标 isOpen: true 游标 isConsumed: false 处理完成总数 1000000 游标 isConsumed: true如果isConsumed在遍历过程中一直是false遍历结束后变true说明游标是逐条消费的。如果一开始就true那多半是驱动把结果全读进来了回去检查useCursorFetch和fetchSize。5. 本篇常见错排查5.1 报错A Cursor is already closed这个报错几乎都跟事务边界有关。Transactional方法返回后事务提交SqlSession关闭游标随之失效。如果你把Cursor作为返回值传出方法外调用方再遍历就会撞上这个错。正确做法是在事务方法内部完成全部消费或者改用手动SqlSession并自己控制commit和close时机。5.2 fetchSize 不生效内存照样涨MySQL 驱动下fetchSize生效的前提是连接串带useCursorFetchtrue。只设Options(fetchSize 1000)而不改连接串驱动仍然走全量加载。另外resultSetType必须是FORWARD_ONLY设成SCROLL_INSENSITIVE会让驱动缓存结果集。5.3 游标遍历中途连接断开流式查询期间数据库连接必须保持打开。如果连接池的max-lifetime小于查询耗时连接会被回收游标中断。把max-lifetime调大或者给流式查询单独配一个连接池。同时default-statement-timeout也要放大默认值往往撑不住长时间游标。5.4 同类内部调用导致事务失效前面提过Transactional基于代理同类内部this.method()调用不走代理事务不生效游标可能提前关闭。解决办法是把流式方法抽到独立的 Service 里或者用AopContext.currentProxy()拿到代理对象再调用。5.5 处理速度慢想加多线程流式查询本身是单连接顺序拉取加多线程要谨慎。常见做法是游标逐条读读到一批后投递到线程池异步处理读和处理解耦。但注意数据库连接不能跨线程共享消费逻辑里不要再碰同一个SqlSession。如果任务偏重、需要长期跑编码辅助或 Agent 类工作可以考虑用 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 把额度固定下来避免调试中途断供。6. 接入与排障的下一步流式查询的配置骨架和验证动作到这里就完整了。回顾一下关键动作连接串加useCursorFetchtrueMapper 返回Cursor并设FORWARD_ONLYService 在事务内完成消费用ResultHandler或forEach逐条处理最后用堆内存对比确认效果。如果你在接入过程中遇到 Key 或通道问题直接去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 检查凭证状态接入细节翻文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。想先确认模型通道是否正常用模型对话 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 发一条消息即可。长期做编码和 Agent 任务的话Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 的额度模型更适合持续使用。最后留一个实用技巧流式查询的fetchSize不是越大越好。设成 1000 到 5000 之间通常比较平衡太小会增加网络往返次数太大又会让单批内存上升。你可以用第 4 节的对比代码把fetchSize分别设成 100、1000、5000 各跑一遍观察内存和耗时的曲线找到适合自己数据特征的那个值。