★ FEATURED ARTICLE 1. 为什么要在存储过程里用游标逐行处理MySQL 游标CURSOR是存储过程/函数里用来「逐行」读取结果集的机制。平时我们写 SQL 都是集合操作一条 UPDATE 影响几千行但有些场景必须一行一行来比如每行数据要单独调用一次外部接口、每行要按不同规则写日志、每行要触发不同的业务分支。这时候游标就派上用场了。我最近在做一个数据清洗任务需要把一张表里的每条记录逐条送去 AI 做语义分类再把结果写回另一张表。集合操作搞不定这种「一行一次外部调用」的逻辑于是用存储过程 游标 TaoToken 统一 API 通道来实现。TaoToken 在这里的作用是把模型调用收敛到一个 Key、一个 API 地址数据库侧只需要发 HTTP 请求不用在存储过程里散落多家厂商的密钥和地址。这篇聚焦三件事DECLARE CURSOR 的完整骨架声明、OPEN、FETCH、CLOSE、HANDLER 异常处理、TaoToken 在数据库侧调用的配置方式、以及用小表 SELECT ... LIMIT 验证逐行处理结果的执行步骤。适合已经会写基础存储过程、但游标老是踩坑的同学。2. TaoToken 前置统一 Key 与 API 通道在存储过程里调 AI最怕的是密钥硬编码、地址到处改。TaoToken 的思路是给你一个统一的 API 入口和一把 Key模型切换、额度管理都在控制台完成数据库侧只认一个地址。你需要先拿到两样东西API Key在控制台的 API Keys 页面创建形如sk-xxxx只显示一次记得存好。API 地址https://taotoken.net/api这是所有请求的基地址。控制台入口在这里https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite创建 Key 的页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite注意Key 不要写死在存储过程源码里。生产环境建议放到独立的配置表或环境变量存储过程通过变量读取。本文为了演示清晰用会话变量传入。如果你只是想先验证模型能不能通可以先用模型对话页面手动发一条消息确认 Key 有效https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite接入文档请求格式、返回结构、错误码在这里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite3. 可复制配置游标骨架 TaoToken 调用3.1 游标完整骨架先看最核心的游标结构。下面这段是标准写法声明、HANDLER、OPEN、FETCH、WHILE、CLOSE 一个不少DELIMITER $$ CREATE PROCEDURE proc_cursor_demo() BEGIN -- 1. 声明变量必须放在游标声明之前 DECLARE v_id INT DEFAULT NULL; DECLARE v_content VARCHAR(500) DEFAULT NULL; DECLARE v_done TINYINT DEFAULT 0; -- 2. 声明游标 DECLARE cur_rows CURSOR FOR SELECT id, content FROM source_table WHERE status 0 LIMIT 100; -- 3. 声明 NOT FOUND 处理器必须在游标之后 DECLARE CONTINUE HANDLER FOR NOT FOUND SET v_done 1; -- 4. 打开游标 OPEN cur_rows; -- 5. 循环读取 read_loop: LOOP FETCH cur_rows INTO v_id, v_content; IF v_done 1 THEN LEAVE read_loop; END IF; -- 这里写逐行处理逻辑 -- 例如调用外部接口、写日志、更新状态 INSERT INTO process_log(id, note) VALUES (v_id, CONCAT(processed: , v_content)); END LOOP; -- 6. 关闭游标 CLOSE cur_rows; END$$ DELIMITER ;几个容易踩的点我按顺序说变量声明必须在游标之前游标声明必须在 HANDLER 之前顺序错了直接报语法错误。HANDLER 用NOT FOUND而不是SQLSTATE 02000也行两者等价NOT FOUND可读性更好。v_done标志位是必须的因为 FETCH 到末尾不会自动退出循环只会触发 HANDLER。3.2 游标条件动态化如果游标里的 WHERE 条件要随参数变化做法是在游标定义时用变量占位OPEN 之前给变量赋值。DELIMITER $$ CREATE PROCEDURE proc_cursor_dynamic(IN p_status INT) BEGIN DECLARE v_id INT DEFAULT NULL; DECLARE v_content VARCHAR(500) DEFAULT NULL; DECLARE v_done TINYINT DEFAULT 0; DECLARE v_status INT DEFAULT 0; -- 用变量承接参数 SET v_status p_status; DECLARE cur_rows CURSOR FOR SELECT id, content FROM source_table WHERE status v_status LIMIT 100; DECLARE CONTINUE HANDLER FOR NOT FOUND SET v_done 1; OPEN cur_rows; read_loop: LOOP FETCH cur_rows INTO v_id, v_content; IF v_done 1 THEN LEAVE read_loop; END IF; -- 处理逻辑 END LOOP; CLOSE cur_rows; END$$ DELIMITER ;注意MySQL 游标不支持在 OPEN 之后重新绑定条件。要换条件只能 CLOSE 再重新 OPEN或者用动态 SQLPREPARE/EXECUTE拼语句。后者更灵活但更容易出注入问题参数一定要用USING传。3.3 TaoToken 调用配置骨架存储过程本身不能直接发 HTTP 请求MySQL 也没有内置的 HTTP 客户端。常见做法有两种一是用 MySQL 的 UDF如lib_mysqludf_http扩展二是把逐行数据导出后由外部程序处理。这里给一个「存储过程负责逐行取数、外部脚本负责调 TaoToken」的分工骨架这是最稳的。存储过程只做取数和落表DELIMITER $$ CREATE PROCEDURE proc_export_for_ai() BEGIN DECLARE v_id INT DEFAULT NULL; DECLARE v_content VARCHAR(500) DEFAULT NULL; DECLARE v_done TINYINT DEFAULT 0; DECLARE cur_rows CURSOR FOR SELECT id, content FROM source_table WHERE ai_status 0 LIMIT 50; DECLARE CONTINUE HANDLER FOR NOT FOUND SET v_done 1; OPEN cur_rows; read_loop: LOOP FETCH cur_rows INTO v_id, v_content; IF v_done 1 THEN LEAVE read_loop; END IF; INSERT INTO ai_task_queue(source_id, payload, created_at) VALUES (v_id, v_content, NOW()); UPDATE source_table SET ai_status 1 WHERE id v_id; END LOOP; CLOSE cur_rows; END$$ DELIMITER ;外部脚本读取ai_task_queue逐条调 TaoTokenimport requests import pymysql API_URL https://taotoken.net/api/v1/chat/completions API_KEY sk-你的Key conn pymysql.connect(host127.0.0.1, userroot, passwordxxx, databasedemo) cur conn.cursor(pymysql.cursors.DictCursor) cur.execute(SELECT id, payload FROM ai_task_queue WHERE done 0 LIMIT 50) for row in cur.fetchall(): resp requests.post( API_URL, headers{Authorization: fBearer {API_KEY}}, json{ model: gpt-4o-mini, messages: [ {role: user, content: f给这段文本分类{row[payload]}} ] }, timeout30 ) result resp.json()[choices][0][message][content] cur.execute( UPDATE ai_task_queue SET result %s, done 1 WHERE id %s, (result, row[id]) ) conn.commit()这样分工的好处存储过程专注逐行取数外部脚本专注网络调用两边解耦。TaoToken 的 Key 只出现在脚本里不进数据库源码。4. 验证请求与成功结果4.1 用小表验证游标逐行处理别一上来就跑全表。先建一张 5 行的小表验证游标逻辑对不对CREATE TABLE source_table ( id INT PRIMARY KEY AUTO_INCREMENT, content VARCHAR(500), status INT DEFAULT 0, ai_status INT DEFAULT 0 ); INSERT INTO source_table(content, status) VALUES (第一条测试数据, 0), (第二条测试数据, 0), (第三条测试数据, 0), (第四条测试数据, 0), (第五条测试数据, 0);调用存储过程CALL proc_export_for_ai();验证结果SELECT * FROM ai_task_queue; SELECT id, ai_status FROM source_table;预期看到ai_task_queue里有 5 条记录source_table的ai_status全部变成 1。如果条数不对说明游标循环有问题先查 HANDLER 和v_done标志位。4.2 验证 TaoToken 通道先用 curl 确认 Key 和地址通curl -X POST https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer sk-你的Key \ -H Content-Type: application/json \ -d { model: gpt-4o-mini, messages: [{role: user, content: 回复OK}] }返回里能看到choices[0].message.content就说明通道正常。这一步过了再跑 Python 脚本。4.3 验证逐行处理结果跑完脚本后检查每条记录是否都有对应的 AI 结果SELECT q.id, q.payload, q.result, q.done FROM ai_task_queue q WHERE q.done 1 ORDER BY q.id;如果某条result为空但done 1说明脚本里没做异常捕获接口报错也标了完成。建议在脚本里加 try/except失败的不标 done下次重跑。5. 本篇常见错排查报错 1329No data - zero rows fetched这是 FETCH 到末尾后没有正确退出循环。检查 HANDLER 是否声明在游标之后v_done是否在 FETCH 之后立刻判断。顺序错了就会一直循环或直接报错。报错 1337Cursor is not openCLOSE 了一个没 OPEN 的游标或者 OPEN 失败后仍然执行了 CLOSE。建议在 OPEN 和 CLOSE 之间加异常处理确保成对出现。游标只处理了第一行多半是 FETCH 只写了一次或者 WHILE 条件写错。标准写法是 LOOP FETCH IF done THEN LEAVEFETCH 必须在循环体内。HANDLER 把其他错误也吞了CONTINUE HANDLER FOR NOT FOUND只捕获「无数据」这一种情况不会吞其他错误。但如果你写成FOR SQLEXCEPTION那所有异常都会被捕获调试时很难发现问题。建议调试阶段先不加 SQLEXCEPTION 处理器。TaoToken 返回 401Key 无效或没带Bearer前缀。检查请求头格式Authorization: Bearer sk-xxx中间有一个空格。TaoToken 返回 429请求太频繁。逐行调用时建议加time.sleep(0.5)控制节奏或者用批量接口一次处理多条。存储过程里变量为 NULL 导致逻辑跳过游标 FETCH 到的字段如果是 NULLIF v_content IS NOT NULL这类判断会直接跳过。要么在 SQL 里用IFNULL兜底要么在逻辑里显式处理 NULL。6. 继续深入Coding Plan 与接入文档如果你要把这套「游标逐行 AI 处理」的模式用到长期跑的数据管道或 Agent 任务里单次调用按量计费可能不够划算。TaoToken 的 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_campaignrewriteKey 管理和额度查看在控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite最后给一个我踩过的坑游标里做 UPDATE 时如果更新的就是游标正在读的那张表MySQL 可能会锁行甚至死锁。稳妥做法是游标只读、把结果写到队列表更新操作放到循环外或由外部脚本完成。这样既避免了锁竞争也让逐行处理的结果可追溯、可重跑。
阅读完成 · 觉得有帮助？
