Spring Batch(7)——数据库批数据读写：TaoToken 统一 Key 接入与分片事务验证

1. 从一次分片读写翻车说起Spring Batch 数据库批数据读写到底难在哪Spring Batch 做数据库批数据读写听起来就是「读一张表、写另一张表」但真到分片加事务的场景坑比扁平文件多得多。我见过太多项目在单机小数据量下跑得好好的一上分片就出现重复写入、回滚不干净、重试后数据翻倍的问题。核心检索词先摆出来Spring Batch 数据库批数据读写指的是用JdbcPagingItemReader分页读库、用JdbcBatchItemWriter批量写库并通过 chunk 事务边界控制提交与回滚的一整套机制。它能帮你把百万级数据的迁移、对账、清洗任务拆成可控的小批次适合谁适合已经在用 Spring Boot 做业务、需要定时跑批或做数据同步的后端同学。为什么数据库读写比文件读写更微妙文件写入时框架要维持文件句柄打开、异常时擦除已写内容所以 Spring Batch 提供了完整的FlatFileItemWriter。但数据库不一样连接池本身保证「连接-写入-释放」的高效数据库驱动自带事务能力异常时自动回滚不存在「擦除半截文件」的问题。所以 Spring Batch 干脆不提供数据库写入实现类把ItemWriter交给开发者自己写。这既是自由也是责任——你得自己保证批量写入和 chunk 事务对齐。分页读取这边JdbcPagingItemReader每次从库里捞一整页但对外仍然一行一行返回。框架根据运行情况决定何时执行下一页查询。这里有个关键点分页查询依赖sortKey排序键如果排序键不唯一翻页时可能漏读或重读。我踩过的坑就是拿一个可重复的字段当 sortKey结果第 2 页和第 1 页出现同一条记录。所以 sortKey 一定要选主键或唯一索引列。再叠加一个现实问题很多团队的批处理任务需要调用外部模型服务做数据补全或校验比如给每条记录打标签、做语义去重。这时候鉴权散落在各个 Job 里Key 管理混乱。本篇会把 TaoToken 统一 Key 接入作为鉴权通道让批处理任务在读取和写入之外多一步可控的外部调用同时用最小数据集验证分片读写和回滚行为。下面从环境准备开始一步步跑通。2. TaoToken 统一 Key 前置准备批处理任务鉴权通道怎么接在动手写 Job 之前先把鉴权通道理清楚。批处理任务里如果每条记录都要调一次外部接口Key 的注入方式直接决定你后面排障的难度。TaoToken 在这里扮演的角色是统一 Key 和 API 通道你不需要在代码里硬编码多个服务的凭证而是通过一个 Base URL 加一个 Key走 OpenAI 兼容协议完成调用。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 注意 API 地址不带 UTM 参数。先说清楚它不是什么它不是数据库代理也不替代你的 DataSource。它只负责外部模型调用的鉴权与转发。你的 Spring Batch 任务依然直连自己的 MySQL/PostgreSQL读写逻辑不变只是在 Processor 阶段需要调用模型时走 TaoToken 的通道。第一步拿到 Key。进入控制台创建 API Key地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。创建后复制保存Key 只显示一次。这里建议给批处理任务单独建一个 Key方便按任务维度排查调用量也方便出问题时单独吊销不影响其他业务。第二步确认你要用的模型 ID。批处理里做文本处理常用的是对话类模型。你可以先在模型对话页面试一下地址是 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 确认模型可用、返回格式符合预期再写进配置。不要凭记忆填模型名模型 ID 写错会直接报 404 或 model not found。第三步把配置写进 Spring Boot 的application.yml。这里给出可复制的片段路径和字段名按你项目实际情况调整taotoken: base-url: https://taotoken.net/api api-key: ${TAOTOKEN_API_KEY} model-id: your-model-id connect-timeout: 5000 read-timeout: 30000注意api-key用环境变量注入不要明文写进仓库。批处理任务通常在服务器上跑环境变量在启动脚本里 export 即可。read-timeout给 30 秒因为批处理里模型调用可能比在线请求慢超时太短会导致 chunk 频繁失败重试。第四步如果你用的是 Claude Code 这类编码工具来辅助写批处理代码可以走 coding-plan 通道地址是 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。它适合长期编码场景和本篇的运行时鉴权是两回事别混用。运行时批处理任务用的是 API Key编码辅助用的是另一套额度。前置准备的核心就一句话Base URL、Key、Model ID 三件套齐全且 Key 通过环境变量注入。后面所有配置都围绕这三件套展开。如果你在接入文档里看到字段名和这里不一致以文档为准地址是 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。3. 可复制配置JdbcPagingItemReader 分片读取与 JdbcBatchItemWriter 批量写入这一节是全文的技术核心给出能直接抄的配置。先建两张表源表和目标表结构一致方便核对行数。DDL 如下CREATE TABLE src_weather ( id int(10) unsigned NOT NULL AUTO_INCREMENT, siteid varchar(64) NOT NULL, month varchar(64) NOT NULL, type varchar(64) NOT NULL, value int(11) NOT NULL, ext varchar(255) DEFAULT NULL, PRIMARY KEY (id) ); CREATE TABLE dst_weather ( id int(10) unsigned NOT NULL AUTO_INCREMENT, siteid varchar(64) NOT NULL, month varchar(64) NOT NULL, type varchar(64) NOT NULL, value int(11) NOT NULL, ext varchar(255) DEFAULT NULL, PRIMARY KEY (id) );插入 1000 条测试数据到src_weather用存储过程或批量 insert 都行。数据量不用大1000 条足够验证分片和回滚。接下来是JdbcPagingItemReader的配置。分页读取的关键是PagingQueryProvider不同数据库分页语法不同用SqlPagingQueryProviderFactoryBean自动识别Bean public SqlPagingQueryProviderFactoryBean queryProvider(DataSource dataSource) { SqlPagingQueryProviderFactoryBean provider new SqlPagingQueryProviderFactoryBean(); provider.setDataSource(dataSource); provider.setSelectClause(select id, siteid, month, type, value, ext); provider.setFromClause(from src_weather); provider.setWhereClause(where id :startId); provider.setSortKey(id); return provider; } Bean public ItemReaderWeatherEntity jdbcPagingItemReader( DataSource dataSource, PagingQueryProvider queryProvider, RowMapperWeatherEntity rowMapper) { MapString, Object parameterValues new HashMap(); parameterValues.put(startId, 0); return new JdbcPagingItemReaderBuilderWeatherEntity() .name(weatherPagingReader) .dataSource(dataSource) .queryProvider(queryProvider) .parameterValues(parameterValues) .rowMapper(rowMapper) .pageSize(200) .saveState(true) .build(); }pageSize设 200配合 chunk 的 50意味着每读 4 个 chunk 才翻一页。saveState(true)让读取位置写入ExecutionContext任务重启时能从上次位置继续而不是从头再来。sortKey必须是唯一列这里用主键id。然后是JdbcBatchItemWriter。Spring Batch 不提供数据库写入实现但JdbcBatchItemWriter是官方提供的批量写入类底层用JdbcTemplate.batchUpdateBean public ItemWriterWeatherEntity jdbcBatchItemWriter(DataSource dataSource) { return new JdbcBatchItemWriterBuilderWeatherEntity() .dataSource(dataSource) .sql(INSERT INTO dst_weather(siteid, month, type, value, ext) VALUES (:siteId, :month, :type, :value, :ext)) .beanMapped() .build(); }注意这里用的是命名参数:siteId而不是问号占位符配合beanMapped()自动从实体属性取值。如果你用问号占位符需要改成itemPreparedStatementSetter。两种方式都行命名参数可读性更好。Step 和 Job 的配置重点是 chunk 大小和事务边界Bean public Step dbToDbStep(StepBuilderFactory builder, ItemReaderWeatherEntity reader, ItemWriterWeatherEntity writer) { return builder.get(dbToDbStep) .WeatherEntity, WeatherEntitychunk(50) .reader(reader) .writer(writer) .faultTolerant() .skipLimit(10) .skip(FlatFileParseException.class) .retryLimit(3) .retry(TransientDataAccessException.class) .build(); } Bean public Job dbToDbJob(JobBuilderFactory builder, Step dbToDbStep) { return builder.get(dbToDbJob) .start(dbToDbStep) .build(); }chunk(50)是事务边界每 50 条提交一次。retryLimit(3)配合retry(TransientDataAccessException.class)让瞬时数据库异常自动重试重试时整个 chunk 回滚重来。这里有个细节重试是针对 chunk 的不是针对单条记录所以 Writer 必须保证幂等否则重试会导致重复写入。JdbcBatchItemWriter的 insert 本身不幂等如果你的业务要求幂等需要在 SQL 里加ON DUPLICATE KEY UPDATE或先删后插。如果你在 Processor 阶段要调 TaoToken 做数据补全配置如下Bean public ItemProcessorWeatherEntity, WeatherEntity enrichProcessor( Value(${taotoken.base-url}) String baseUrl, Value(${taotoken.api-key}) String apiKey, Value(${taotoken.model-id}) String modelId) { return item - { // 调用 TaoToken 通道做数据补全 String enriched callModel(baseUrl, apiKey, modelId, item.getExt()); item.setExt(enriched); return item; }; }callModel用 RestTemplate 或 WebClient 发 OpenAI 兼容请求Header 里带Authorization: Bearer ${apiKey}。注意 Processor 在 chunk 事务内执行如果模型调用超时抛异常整个 chunk 回滚。所以模型调用的超时和重试策略要和 chunk 的 retry 配置对齐避免一个慢请求拖垮整个批次。4. 验证请求与成功结果最小数据集跑通分片读写并核对行数配置写完跑一次最小数据集验证。启动 Job 的方式有两种命令行--spring.batch.job.namesdbToDbJob或者写个CommandLineRunner手动触发。这里用命令行方便观察日志。启动后日志里会看到几个关键节点。第一JdbcPagingItemReader初始化打印pageSize200。第二chunk 开始执行每 50 条一次Transaction committed。第三Step 结束时打印StepExecution的读写计数。跑完后核对行数执行SELECT COUNT(*) FROM src_weather; SELECT COUNT(*) FROM dst_weather;如果源表 1000 条目标表也应该是 1000 条。如果目标表少于 1000说明有 chunk 失败被跳过或回滚了。如果多于 1000说明重试导致重复写入需要检查 Writer 幂等性。再验证分片读取的翻页行为。把日志级别调到 DEBUG搜索JdbcPagingItemReader的 SQL 执行记录应该看到类似Executing SQL: SELECT id, siteid, month, type, value, ext FROM src_weather WHERE id ? ORDER BY id ASC LIMIT 200每次翻页id ?的参数会变成上一页最后一条的 id。如果参数没变说明saveState或sortKey配置有问题会导致死循环或漏读。验证回滚行为故意制造异常。在 Writer 里加一段逻辑当id 500时抛RuntimeException。重新跑 Job观察日志。你会看到第 500 条所在的 chunk假设是第 10 个 chunk覆盖 451-500整体回滚dst_weather里不会有 451-500 这 50 条。同时由于skipLimit(10)和skip(FlatFileParseException.class)只跳过特定异常RuntimeException不在跳过列表里所以 Job 会失败退出。这正好验证了事务边界chunk 内要么全成功要么全回滚。如果你想验证重试把异常改成TransientDataAccessException的子类比如QueryTimeoutException。配置了retryLimit(3)后框架会重试这个 chunk 最多 3 次。如果 3 次都失败Job 失败如果第 2 次成功Job 继续。重试时整个 chunk 回滚重来所以 Writer 的幂等性在这里至关重要。最后验证 TaoToken 鉴权通道。在 Processor 里调一次模型观察返回。如果 Key 正确、模型 ID 正确会拿到正常响应。如果 Key 错误会收到 401。如果 Base URL 写错会收到连接超时或 404。这一步的验证结果直接决定后面排障的方向。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 报错对照批处理任务跑起来后报错集中在几个地方。这一节按真实报错对照排查。401 Unauthorized。这是 TaoToken 鉴权失败。检查三件事Key 是否通过环境变量正确注入Header 里是否带了Authorization: Bearer keyKey 是否被吊销。常见错误是把 Key 写进了application.yml但没加${}或者环境变量名拼错。排查方法在启动日志里打印 Key 的前 6 位和后 4 位确认注入成功。不要打印完整 Key。local proxy failed / connection refused。这是网络层问题。检查 Base URL 是否写成了https://taotoken.net/api注意结尾没有斜杠。如果你在容器里跑检查容器是否能访问外网。如果你配置了 HTTP 代理检查代理是否放行了taotoken.net。注意这里说的代理是网络代理配置不是让你去搞什么特殊通道企业内网出口代理是正常运维配置。reading choices 报错 / choices 字段为空。这是模型返回格式解析失败。常见原因是模型 ID 写错返回了非预期结构或者请求体里messages格式不对。检查请求体是否符合 OpenAI 兼容格式{ model: your-model-id, messages: [{role: user, content: hello}] }如果返回体里没有choices字段先打印完整响应体看结构。不要盲目按choices[0].message.content取值先确认字段存在。OAuth 相关报错。如果你用的是 Claude Code 或 Codex 这类工具可能会遇到 OAuth 认证问题。注意区分运行时批处理任务用的是 API Key不是 OAuth。OAuth 是编码工具登录用的。如果你在批处理代码里误用了 OAuth token会报 401 或 invalid token。排查方法确认批处理任务用的是api-key配置项不是oauth-token。Codex auth.json 配置问题。如果你用 Codex 辅助写代码auth.json里需要填 Base URL、Key、Model ID 三件套。路径通常在~/.codex/auth.json。配置片段{ base_url: https://taotoken.net/api, api_key: your-key, model: your-model-id }注意base_url不要带/v1除非文档明确要求。三件套缺一不可缺 Key 报 401缺 Model ID 报 404。CC Switch / Cline MCP 配置问题。如果你用 CC Switch 或 Cline 的 MCP 功能同样需要 Base URL、Key、Model ID 三件套。MCP 配置里通常写在mcp.json或工具设置里。常见错误是把 MCP 直连到生产数据库这是禁止的。MCP 只用于编码辅助不要让它碰生产库。分页读取漏读或重读。检查sortKey是否唯一。如果 sortKey 有重复值翻页时id ?的条件会跳过或重复记录。解决方法sortKey 用主键或者用「主键 唯一列」组合排序。chunk 重试导致重复写入。检查 Writer 是否幂等。JdbcBatchItemWriter的 insert 不幂等重试会重复插入。解决方法SQL 改成INSERT ... ON DUPLICATE KEY UPDATE或者在 Writer 里先按业务键删除再插入。事务边界不清导致部分提交。检查 chunk 大小和faultTolerant配置。如果skip配置了宽泛的异常类型被跳过的记录不会回滚会导致数据不一致。建议skip只配置明确的业务异常不要用Exception.class一把梭。6. 继续把批处理跑稳从验证到长期运行的接入建议跑通一次最小数据集只是开始长期运行还要考虑几件事。第一Key 的轮换。批处理任务通常长期跑Key 泄露风险高。建议定期在控制台轮换 Key地址是 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 轮换后更新环境变量并重启任务。第二调用量监控。批处理任务调用量大建议在 TaoToken 控制台按 Key 维度看用量避免超额。第三失败告警。Job 失败时要有告警不要等第二天才发现数据没同步。如果你还在选型阶段想先验证模型返回是否符合预期可以去模型对话页面试几条地址是 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。确认没问题再写进批处理配置。如果你需要长期编码辅助来维护这套批处理代码可以看 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 字段名和参数以文档为准。最后给一个实用技巧批处理任务的日志里把 chunk 的读写计数、翻页 SQL、模型调用耗时都打出来。出问题时先看计数对不对再看翻页参数变没变最后看模型调用有没有超时。这三步能定位 80% 的问题。剩下的 20%多半是事务边界和幂等性回到第 3 节的配置逐项核对即可。