SpringBatch 集成 MyBatis 循环批量读取 MySQL：TaoToken 统一 Key 配置与分页读取验证

1. SpringBatch 每次调用 Job 都读不到新数据问题到底出在哪如果你正在做批处理数据同步大概率会遇到这个场景服务启动后第一次手动触发 Job数据正常处理第二次再触发日志显示Job already completed或者 Step 直接跳过Reader 里一条数据都没有。这不是 SpringBatch 的 bug而是它的设计逻辑——JobInstance 由 jobName jobParameters 唯一确定参数没变就认为是同一个实例不会重复执行。我试过最典型的翻车现场把查询条件写死在param里createDate永远是同一天RunIdIncrementer又没加结果第二次调用直接被JobRepository拦下。还有一种更隐蔽的情况Reader 用的是MyBatisPagingItemReader但pageSize和chunkSize没对齐第一轮读完page0之后第二轮page1查出来是空列表循环提前终止你以为读完了其实只处理了第一页。这篇要解决的核心问题就三个让每次 Job 调用都重新从 MySQL 拉最新列表、用游标方式避免分页错位、把每轮读取条数和终止条件用日志打出来可验证。适合正在用 SpringBatch MyBatis 做订单同步、用户标签刷新、日志归档这类批处理任务的同学。下面从 TaoToken 的前置配置开始一步步把可复制的骨架搭出来。2. TaoToken 统一 Key 前置配置settings.json 与 config.toml在写 Job 之前先把模型调用的 Key 统一管起来。批处理服务里经常会有多个模块要调模型做数据清洗、字段补全、异常分类如果每个模块各自维护一套 Key后面换 Key 就是灾难。TaoToken 的做法是提供一个统一的 API 入口你只需要在配置文件里写一次。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 注意 API 地址不带 UTM 参数直接填就行。先看settings.json这个文件一般放在项目根目录或者~/.config下用于声明默认的模型和 Key 引用{ default_model: claude-sonnet-4-20250514, api_base: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, timeout_seconds: 60, max_retries: 3, batch: { chunk_size: 10000, page_size: 1000, log_every_n_chunk: 1 } }再看config.toml这个更适合放在 SpringBoot 的resources目录下用ConfigurationProperties读进来[taotoken] api-base https://taotoken.net/api api-key ${TAOTOKEN_API_KEY} default-model claude-sonnet-4-20250514 connect-timeout 10000 read-timeout 60000 [taotoken.batch] chunk-size 10000 page-size 1000 max-item-count 500000注意api-key不要硬编码在文件里用环境变量注入。SpringBoot 启动时通过-DTAOTOKEN_API_KEYxxx或者系统环境变量传入避免提交到 Git。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 。如果你后面要做长期编码或者 Agent 任务可以看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。3. 可复制配置Job/Step 骨架 MyBatis 游标 Reader3.1 为什么选 MyBatisCursorItemReader 而不是 PagingItemReaderMyBatisPagingItemReader依赖page和pageSize两个参数每次read()触发一次LIMIT offset, size查询。问题在于如果查询期间数据被修改offset 会漂移导致漏读或重复读。而MyBatisCursorItemReader底层用的是 JDBC 游标一次查询建立结果集逐条next()读取不会因为数据变动导致分页错位而且内存占用稳定。代价是游标需要保持数据库连接打开长事务要注意连接池配置。对于批处理场景这个代价可以接受。3.2 BatchConfig 里初始化 SqlSessionFactory先改造基础配置类把SqlSessionFactory做成单例避免每次 Step 都重新构建import org.apache.ibatis.session.SqlSessionFactory; import org.mybatis.spring.SqlSessionFactoryBean; import org.mybatis.spring.boot.autoconfigure.MybatisProperties; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.io.Resource; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import javax.sql.DataSource; import java.util.concurrent.ThreadPoolExecutor; Configuration public class BatchConfig { Autowired protected JobBuilderFactory jobBuilderFactory; Autowired protected StepBuilderFactory stepBuilderFactory; Autowired private MybatisProperties mybatisProperties; Autowired private DataSource dataSource; private static volatile SqlSessionFactory sessionFactory; Bean public ThreadPoolTaskExecutor batchTaskExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); executor.setCorePoolSize(10); executor.setMaxPoolSize(15); executor.setKeepAliveSeconds(300); executor.setQueueCapacity(10000); executor.setThreadNamePrefix(spring_batch_); executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy()); executor.initialize(); return executor; } public SqlSessionFactory sqlSessionFactory() { if (sessionFactory null) { synchronized (BatchConfig.class) { if (sessionFactory null) { try { SqlSessionFactoryBean bean new SqlSessionFactoryBean(); bean.setDataSource(dataSource); Resource[] mapperLocations mybatisProperties.resolveMapperLocations(); if (mapperLocations ! null) { bean.setMapperLocations(mapperLocations); } sessionFactory bean.getObject(); } catch (Exception e) { throw new IllegalStateException(SqlSessionFactory 初始化失败, e); } } } } return sessionFactory; } }这里用双重检查锁保证单例volatile防止指令重排。resolveMapperLocations()会读取mybatis.mapper-locations配置通常是classpath*:mapper/**/*.xml。3.3 封装游标 Reader 工具类import org.apache.ibatis.session.SqlSessionFactory; import org.mybatis.spring.batch.MyBatisCursorItemReader; import org.springframework.batch.item.ItemReader; import java.util.Map; public class MyBatisReaderFactory { public static T ItemReaderT cursorReader(MapString, Object params, SqlSessionFactory sqlSessionFactory) { MyBatisCursorItemReaderT reader new MyBatisCursorItemReader(); reader.setSqlSessionFactory(sqlSessionFactory); reader.setParameterValues(params); Object queryId params.get(queryId); if (queryId null) { throw new IllegalArgumentException(params 中必须包含 queryId); } reader.setQueryId(queryId.toString()); return reader; } }关键点queryId必须是 Mapper 接口的全限定名 方法名比如com.example.batch.dao.UserDao.selectByDate。parameterValues里除了queryId其他键值会作为 MyBatis 的参数传入。3.4 Job 与 Step 配置import org.springframework.batch.core.Job; import org.springframework.batch.core.Step; import org.springframework.batch.core.launch.support.RunIdIncrementer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import javax.annotation.Resource; import java.util.HashMap; import java.util.Map; Configuration public class UserBatchConfig extends BatchConfig { Resource private UserDao userDao; Bean public Step userSyncStep() { MapString, Object params new HashMap(); params.put(queryId, com.example.batch.dao.UserDao.selectByDate); params.put(createDate, 2025-01-01); return stepBuilderFactory.get(userSyncStep) .User, Userchunk(10000) .reader(MyBatisReaderFactory.cursorReader(params, sqlSessionFactory())) .processor(new UserProcessor()) .writer(new MysqlWriter(userDao)) .taskExecutor(batchTaskExecutor()) .throttleLimit(4) .build(); } Bean public Job userSyncJob() { return jobBuilderFactory.get(userSyncJob) .incrementer(new RunIdIncrementer()) .listener(new UserJobCompletionListener()) .start(userSyncStep()) .build(); } }RunIdIncrementer是解决「第二次调用不执行」的关键——它每次运行都会给jobParameters加一个递增的run.id让JobInstance唯一从而绕过「已完成」判断。3.5 Mapper 与 XMLMapper public interface UserDao { Integer add(User user); ListUser selectByDate(MapString, Object map); }?xml version1.0 encodingUTF-8? !DOCTYPE mapper PUBLIC -//mybatis.org//DTD Mapper 3.0//EN http://mybatis.org/dtd/mybatis-3-mapper.dtd mapper namespacecom.example.batch.dao.UserDao insert idadd parameterTypecom.example.batch.entity.User INSERT INTO user_batch_test (id, age, name, create_date) VALUES (#{id}, #{age}, #{name}, #{createDate}) /insert select idselectByDate resultTypecom.example.batch.entity.User parameterTypejava.util.HashMap SELECT id, age, name, create_date FROM user_batch_test WHERE create_date IS NOT NULL if testcreateDate ! null AND create_date #{createDate} /if ORDER BY id /select /mapperORDER BY id很重要游标读取依赖稳定的排序否则结果集顺序不确定会导致处理逻辑出错。4. 验证请求与成功结果日志打点与断点确认4.1 在 Reader 外面包一层日志装饰器SpringBatch 自带的ItemReadListener可以在beforeRead、afterRead、onReadError打点import org.springframework.batch.core.ItemReadListener; public class ReadCountListener implements ItemReadListenerUser { private long count 0; Override public void beforeRead() { // 每轮读取前触发 } Override public void afterRead(User item) { count; if (count % 1000 0) { System.out.println([Reader] 已读取条数: count , 当前 id item.getId()); } } Override public void onReadError(Exception ex) { System.err.println([Reader] 读取异常: ex.getMessage()); } public long getCount() { return count; } }挂到 Step 上.reader(MyBatisReaderFactory.cursorReader(params, sqlSessionFactory())) .listener(new ReadCountListener())4.2 用 StepExecution 确认每轮读取条数在afterStep里拿readCount、writeCount、commitCountimport org.springframework.batch.core.StepExecution; import org.springframework.batch.core.annotation.AfterStep; public class StepSummaryListener { AfterStep public void afterStep(StepExecution stepExecution) { System.out.println(Step: stepExecution.getStepName() , readCount stepExecution.getReadCount() , writeCount stepExecution.getWriteCount() , commitCount stepExecution.getCommitCount() , status stepExecution.getStatus()); } }4.3 循环终止条件怎么判断MyBatisCursorItemReader的read()返回null时SpringBatch 认为数据读完Step 进入COMPLETED。所以终止条件就是游标结果集耗尽。你可以在日志里看到readCount和数据库实际条数一致就说明没有漏读。如果发现readCount小于预期检查三个地方SQL 的WHERE条件是否过滤掉了数据、chunkSize是否大于结果集导致一次读完、taskExecutor多线程下throttleLimit是否限制了并发。4.4 手动触发验证用JobLauncher手动触发两次观察第二次是否重新读取Autowired private JobLauncher jobLauncher; Autowired private Job userSyncJob; public void runJob() throws Exception { JobParameters params new JobParametersBuilder() .addLong(timestamp, System.currentTimeMillis()) .toJobParameters(); JobExecution execution jobLauncher.run(userSyncJob, params); System.out.println(Job status: execution.getStatus()); }每次timestamp不同JobInstance就不同Job 会重新执行。日志里应该看到两次完整的readCount输出。5. 本篇常见错排查5.1 第二次调用 Job 显示已完成原因jobParameters没变JobRepository认为实例已存在。解决加RunIdIncrementer或者手动传timestamp参数。注意RunIdIncrementer依赖JobRepository的run.id序列如果用的是内存版MapJobRepository重启后会重置。5.2 Reader 读不到数据readCount0先看 SQL 是否真的能查出数据直接在 MySQL 客户端执行一遍。再看queryId是否写对必须是接口全限定名.方法名。最后看parameterValues里的键名和 XML 里的#{xxx}是否一致大小写敏感。5.3 游标读取报 Connection is closedMyBatisCursorItemReader需要保持连接打开如果dataSource的连接池maxLifetime小于批处理耗时连接会被回收。解决调大maxLifetime或者用HikariCP的keepaliveTime保活。另外chunkSize不要设太大10000 左右比较稳。5.4 多线程下数据重复处理taskExecutor加throttleLimit后多个线程共享同一个 Reader但MyBatisCursorItemReader不是线程安全的。正确做法是每个线程独立 Reader或者用SynchronizedItemStreamReader包装。批处理场景建议单线程读、多线程写读用游标写用taskExecutor。5.5 TaoToken Key 读取失败检查TAOTOKEN_API_KEY环境变量是否注入config.toml里的${TAOTOKEN_API_KEY}占位符是否被 Spring 解析。如果用的是Value(${taotoken.api-key})确保application.yml里有对应配置。Key 本身在控制台可以重新生成地址是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。6. 接入文档与模型验证入口配置和代码都跑通之后建议做两件事一是把 Job 的readCount、writeCount、耗时打到监控里方便排查二是用模型对话验证一下数据清洗逻辑是否符合预期入口在 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 可以直接把一批样本数据贴进去让模型帮你检查字段映射和异常值。完整的接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有 API 参数说明和错误码对照。如果你用的是 Claude Code 做开发Anthropic 兼容配置参考 https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude-code-anthropicutm_campaignrewrite 。最后提醒一句MyBatisCursorItemReader的游标在 MySQL 里默认是FETCH_SIZEInteger.MIN_VALUE也就是逐条从服务器拉。如果数据量大可以在 JDBC URL 里加useCursorFetchtruedefaultFetchSize1000减少网络往返。这个参数我踩过坑不加的话百万级数据读取会慢得离谱。