1. Flask 里那个让人头大的 MySQL Connection not available如果你正在用 Flask 写 Web数据库层用的是mysql-connector-python或者PyMySQL某天突然在日志里看到这么一行mysql.connector.errors.OperationalError: MySQL Connection not available.然后你会发现不只是当前这个请求挂了后面所有涉及数据库的操作全部报同一个错重启服务才能恢复。这个报错在 Flask MySQL 的组合里非常典型尤其是小项目、内部工具、课程作业这类没有专门 DBA 兜底的场景。它到底是什么意思简单说就是你的代码拿着一个「已经不可用」的连接对象去开 cursor 或者执行 SQL。连接不可用的原因有好几类连接池里的连接被 MySQL 服务端超时踢掉了、连接被某个未读完的结果集搞脏了、配置里连的库和实际用的库错位了。这三类问题的表现都是同一句报错但排查路径完全不同。这篇就按「连接池耗尽 / 超时回收 / 配置错位」三个角度拆开讲给你可以直接复制的 Flask SQLAlchemy 连接池参数、一个健康检查路由、以及用 TaoToken 统一 Key 管理多环境配置的settings.json/config.toml骨架。目标很明确让你在十分钟内定位到是哪一类问题并且把连接恢复。适合谁看正在用 Flask 写 Web、数据库连接偶发或必现失败、想搞清楚连接池到底怎么配的开发者。不需要你是 DBA但需要你能改配置文件、能跑一条curl验证。2. 先搞清楚连接为什么会被判定为「不可用」在动手改配置之前得先理解 MySQL 连接的生命周期。一个 TCP 连接建立后MySQL 服务端有一个wait_timeout参数默认 28800 秒8 小时空闲超过这个时间服务端会主动断开。但客户端这边并不知道连接已经断了它还以为连接是好的直到下一次拿这个连接去执行 SQL才会收到「Connection not available」。另一类更隐蔽你打开了一个 unbuffered cursor执行了查询但没有把结果集读完就close()了。MySQL 协议规定未读完的结果集会让这个连接处于「脏」状态后续任何查询都会失败。这就是很多老文章里提到的那个经典 bug解决方式是在关闭前把剩余结果fetch干净。第三类是配置错位你的 Flask 应用连的是 A 库但迁移脚本或者某个后台任务连的是 B 库两边连接池参数不一致导致某一边的连接被大量占用后另一边拿不到可用连接。这三类的共同点是报错信息一样但修复动作完全不同。所以第一步不是急着改代码而是先判断你属于哪一类。2.1 用最小复现确认问题类型写一个最小的 Flask 路由故意制造「连接被服务端断开」的场景# app.py import time from flask import Flask from sqlalchemy import create_engine, text app Flask(__name__) # 故意把 pool_recycle 设得比 MySQL wait_timeout 大 engine create_engine( mysqlpymysql://user:pass127.0.0.1:3306/demo, pool_size5, max_overflow2, pool_recycle3600, # 1 小时回收但服务端可能 10 分钟就断 pool_pre_pingFalse, # 关闭预检模拟「拿到坏连接」 ) app.route(/bad) def bad(): with engine.connect() as conn: conn.execute(text(SELECT 1)) time.sleep(20) # 模拟空闲 with engine.connect() as conn: conn.execute(text(SELECT 1)) return ok把 MySQL 的wait_timeout临时调小比如 10 秒然后访问/bad你大概率会复现Connection not available。这一步的目的是确认你的问题是不是「连接被服务端回收但客户端不知道」。如果复现成功那pool_pre_pingTrue和合理的pool_recycle就是解药。如果复现不出来那问题更可能在「未读完结果集」或「配置错位」上。3. TaoToken 前置把多环境 Key 和连接配置统一管起来排查连接问题时最烦的是配置散落在.env、config.py、docker-compose.yml好几个地方改一个参数要翻三个文件。我习惯把模型调用的 Key 和数据库连接配置都收拢到一份配置骨架里用 TaoToken 统一管理 Key这样本地、测试、生产三套环境的差异只体现在一份文件里。TaoToken 在这里的角色是你通过它拿到统一的 API Key用于模型对话、Coding Plan 等场景同时把数据库连接参数和 Key 放在同一份settings.json或config.toml里避免「Key 在 A 文件、连接串在 B 文件」的错位。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 。你需要先拿到 Key再去控制台确认模型和额度。拿 Key 的入口是 API Keys 页面接入文档在 doc 页面。这两个地址后面 CTA 会再给一次这里先记住Key 拿到后不要硬编码进代码放进配置文件用环境变量注入。3.1 settings.json 骨架{ taotoken: { api_key: ${TAOTOKEN_API_KEY}, base_url: https://taotoken.net/api, default_model: claude-sonnet }, database: { host: 127.0.0.1, port: 3306, user: app_user, password: ${DB_PASSWORD}, name: demo, charset: utf8mb4 }, pool: { pool_size: 5, max_overflow: 10, pool_recycle: 280, pool_pre_ping: true, pool_timeout: 30 } }注意pool_recycle设成 280 秒比 MySQL 默认wait_timeout小很多这样连接在被服务端踢掉之前客户端自己先回收重建。pool_pre_ping打开后每次从池里取连接会先发一个轻量 ping坏连接直接丢弃。3.2 config.toml 骨架如果你更喜欢 TOML[taotoken] api_key ${TAOTOKEN_API_KEY} base_url https://taotoken.net/api default_model claude-sonnet [database] host 127.0.0.1 port 3306 user app_user password ${DB_PASSWORD} name demo charset utf8mb4 [pool] pool_size 5 max_overflow 10 pool_recycle 280 pool_pre_ping true pool_timeout 30两份骨架的字段含义一致选你项目里已经在用的格式就行。关键是数据库连接参数和 TaoToken 的 Key 放在同一份配置里改环境时只改这一份。4. 可复制配置Flask SQLAlchemy 连接池参数怎么填拿到配置骨架后把它接进 Flask。下面是一份可以直接跑的app.py包含连接池参数、健康检查路由和错误处理。# app.py import os import json from flask import Flask, jsonify from sqlalchemy import create_engine, text from sqlalchemy.exc import OperationalError app Flask(__name__) def load_config(pathsettings.json): with open(path, r, encodingutf-8) as f: raw f.read() # 简单替换环境变量占位符 for key, val in os.environ.items(): raw raw.replace(${ key }, val) return json.loads(raw) cfg load_config() db cfg[database] pool cfg[pool] dsn ( fmysqlpymysql://{db[user]}:{db[password]} f{db[host]}:{db[port]}/{db[name]}?charset{db[charset]} ) engine create_engine( dsn, pool_sizepool[pool_size], max_overflowpool[max_overflow], pool_recyclepool[pool_recycle], pool_pre_pingpool[pool_pre_ping], pool_timeoutpool[pool_timeout], futureTrue, ) app.route(/health/db) def health_db(): try: with engine.connect() as conn: conn.execute(text(SELECT 1)) return jsonify({db: ok}), 200 except OperationalError as e: return jsonify({db: fail, error: str(e)}), 503 app.route(/users) def users(): try: with engine.connect() as conn: rows conn.execute(text(SELECT id, name FROM users LIMIT 10)).fetchall() return jsonify([{id: r[0], name: r[1]} for r in rows]) except OperationalError as e: return jsonify({error: db_unavailable, detail: str(e)}), 503几个参数的解释用表格对照一下参数作用建议值踩坑点pool_size池里常驻连接数5–10设太大浪费设太小高并发排队max_overflow超出 pool_size 后临时连接数10设 0 则并发一高就报 pool 满pool_recycle连接回收秒数280必须小于 MySQL wait_timeoutpool_pre_ping取连接前先 pingTrue关掉会拿到坏连接pool_timeout取连接等待秒数30设太小会误报超时pool_recycle是最关键的一个。很多人只改pool_pre_ping但预检本身有开销而且如果连接已经被服务端断开预检只是帮你发现并重建不如主动回收来得干净。两个一起用最稳。4.1 未读完结果集的修复如果你确认问题出在 unbuffered cursor 上修复方式是在close()之前把剩余结果读干净def execute_sql(conn, command, paramsNone, searchTrue): cursor conn.cursor() try: if params is None: cursor.execute(command) else: cursor.execute(command, params) if search: return cursor.fetchall() else: # 关键把剩余结果读干净避免连接变脏 while cursor.fetchone() is not None: pass conn.commit() return cursor.rowcount finally: cursor.close()注意finally里关闭 cursor保证异常时也能释放。while cursor.fetchone()这个循环看起来笨但它是解决「未读完结果集导致连接不可用」最直接的办法。5. 验证请求与成功结果配置改完后先跑健康检查curl -s http://127.0.0.1:5000/health/db正常返回{db: ok}如果返回 503说明连接还是有问题看error字段里的具体信息。接着验证业务路由curl -s http://127.0.0.1:5000/users正常返回用户列表。然后做一次「空闲后恢复」测试等 5 分钟超过你设的pool_recycle再访问一次/users如果还能正常返回说明连接回收和预检生效了。再做一个并发测试确认池参数合理for i in $(seq 1 20); do curl -s http://127.0.0.1:5000/users done wait如果 20 个并发请求都返回正常没有出现QueuePool limit或Connection not available说明pool_size max_overflow够用。如果出现排队超时把max_overflow调大一点。5.1 用 TaoToken 验证模型调用是否正常数据库通了之后顺手验证一下 TaoToken 的 Key 是否可用。用模型对话页面发一条测试消息或者用 API 直接调curl -s https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: claude-sonnet, messages: [{role: user, content: ping}] }返回里有choices字段就说明 Key 正常。这一步和数据库排查是独立的但放在一起做能确认你的配置骨架里两类 Key 都没问题。6. 本篇常见错排查6.1 报错依旧是 Connection not available但 pool_pre_ping 已开检查pool_recycle是不是大于 MySQL 的wait_timeout。用这条 SQL 查服务端超时SHOW VARIABLES LIKE wait_timeout;如果返回 28800而你的pool_recycle是 3600那连接在池里待了 1 小时才回收但服务端 8 小时才断理论上没问题。但如果服务端被改成了 600而你的pool_recycle是 3600那连接在池里待到 600 秒后就被服务端断了客户端却要等到 3600 秒才回收中间这段时间拿到的都是坏连接。把pool_recycle设成比wait_timeout小 20–30 秒。6.2 健康检查通过但业务路由偶发失败大概率是未读完结果集。检查你的代码里有没有cursor.execute()之后只fetchone()一次就close()的地方。把所有非查询类的execute后面补上「读干净」的逻辑或者改用 buffered cursor。6.3 配置错位连的库和预期不一致在健康检查路由里加一行把当前连接的库名打出来with engine.connect() as conn: db_name conn.execute(text(SELECT DATABASE())).scalar() print(connected to:, db_name)如果打出来的库名和你以为的不一样检查settings.json里的database.name以及环境变量DB_PASSWORD有没有被别的 shell 覆盖。6.4 TaoToken Key 报 401先确认TAOTOKEN_API_KEY环境变量在当前 shell 里存在echo $TAOTOKEN_API_KEY如果为空说明配置文件里的${TAOTOKEN_API_KEY}没被替换。检查load_config里的替换逻辑或者直接在启动 Flask 前export一下。Key 的管理入口在 API Keys 页面接入细节看 doc 文档。7. 把连接池和 Key 都收进一份配置里回到最开始的问题MySQL Connection not available之所以难查是因为它把三类不同的问题包装成了同一句报错。你现在手里有了三样东西一份带pool_recycle和pool_pre_ping的连接池配置、一个能区分「连接坏了」和「配置错了」的健康检查路由、一份把数据库参数和 TaoToken Key 放在一起的配置骨架。下一步动作很具体先跑/health/db确认当前状态再对照wait_timeout调pool_recycle最后用并发请求压一下池参数。如果排查过程中需要看模型侧的调用是否正常模型对话页面可以直接发消息验证如果是长期编码或 Agent 场景Coding Plan 页面有对应的套餐说明Key 的创建和轮换在 API Keys 页面接入细节在 doc 文档。把这几步走完连接不可用的问题基本能定位到具体那一类而不是靠重启服务碰运气。