1. 从全局 conn 到连接池Flask 高并发下的真实痛点Flask 里写数据库最容易踩的坑就是把conn和cursor定义成模块级全局变量。单进程调试时看着没问题一旦上 gunicorn 多 worker 或者开多线程多个请求同时操作同一个连接对象轻则结果串台重则直接抛Packet sequence number wrong或者Commands out of sync。我试过在本地用threadedTrue跑一个查询接口20 个并发就把全局 cursor 打崩了。把连接挪进视图函数能解决串台但每次请求都pymysql.connect()再close()高并发下 TCP 握手和 MySQL 认证的开销非常可观。压测 500 QPS 时光是建连就吃掉大量 CPU数据库侧Threads_connected频繁抖动。连接池要解决的就是这两件事连接复用 并发隔离。这篇聚焦 Flask 应用在高并发下数据库连接池的配置与验证同时把 AI 工具侧的接入统一到 TaoToken 的 Key/API 通道上。你会拿到一份可复制的config.toml配置骨架、连接池参数模板以及通过压测和日志验证连接复用、超时回收、连接泄漏的具体动作。适合已经写过 Flask 接口、准备上生产或者正在排查连接数暴涨的开发者。2. TaoToken 前置统一 Key 与 API 通道准备在写连接池代码之前先把 AI 工具侧的接入通道理清楚。TaoToken 提供统一的 Key 和 API 入口Flask 项目里如果要用到模型对话、代码补全这类能力不需要每个工具单独配一套凭证统一走一个 Key 即可。你需要先拿到 API Key。打开控制台创建密钥https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite创建完成后在 API Keys 页面管理你的密钥建议按项目或环境分开建方便后续排查是哪个服务在调用https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteAPI 的基础地址是https://taotoken.net/api注意这个地址不带 UTM 参数直接用于代码里的base_url。如果你在 Flask 里封装 AI 调用客户端把 Key 放进环境变量不要硬编码进config.toml提交到仓库。注意config.toml里只放非敏感的连接池参数和 API 地址Key 一律走环境变量注入。接入文档在这里里面有各语言 SDK 的调用示例Flask 项目可以直接参考 Python 部分https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite如果你后续要做长期编码或者 Agent 类任务可以了解 Coding Plan它更适合持续性的代码生成场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite想先验证模型通道是否通可以直接在模型对话页面发一条测试消息https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite3. 可复制配置config.toml 骨架与连接池参数模板下面这份config.toml是完整的配置骨架包含数据库连接池参数和 AI 通道地址。用tomllibPython 3.11或者tomli读取Flask 启动时加载一次。# config.toml [app] debug false secret_key change-me-in-env [database] host 127.0.0.1 port 3306 user app_user password app_password database app_db charset utf8mb4 [database.pool] # 连接池允许的最大连接数0 和 None 表示不限制 maxconnections 20 # 初始化时至少创建的空闲连接数 mincached 5 # 池中最多闲置的连接数0 和 None 不限制 maxcached 10 # 池中最多共享的连接数pymysql threadsafety1 时该值实际不生效 maxshared 0 # 无可用连接时是否阻塞等待True 等待False 直接报错 blocking true # 单个连接最多被重复使用的次数None 表示无限制 maxusage 1000 # 每次取连接时 ping 服务端0never 1default 2when cursor created 4when query executed 7always ping 4 # 连接空闲回收时间秒超过后从池中移除 maxidle 300 [ai] base_url https://taotoken.net/api # api_key 从环境变量 TAOTOKEN_API_KEY 读取不写在这里 timeout 30读取配置并初始化连接池的代码# settings.py import os import tomllib import pymysql from dbutils.pooled_db import PooledDB with open(config.toml, rb) as f: CFG tomllib.load(f) DB CFG[database] POOL_CFG DB[pool] PYMYSQL_POOL PooledDB( creatorpymysql, maxconnectionsPOOL_CFG[maxconnections], mincachedPOOL_CFG[mincached], maxcachedPOOL_CFG[maxcached], maxsharedPOOL_CFG[maxshared], blockingPOOL_CFG[blocking], maxusagePOOL_CFG[maxusage], pingPOOL_CFG[ping], maxidlePOOL_CFG[maxidle], hostDB[host], portDB[port], userDB[user], passwordDB[password], databaseDB[database], charsetDB[charset], ) TAOTOKEN_API_KEY os.environ.get(TAOTOKEN_API_KEY, ) TAOTOKEN_BASE_URL CFG[ai][base_url]参数对照表方便你按实际负载调整参数作用高并发建议值maxconnections池上限超过则阻塞或报错按 DB 最大连接数的 60% 设置mincached启动时预热连接数5–10避免首波请求建连maxcached闲置连接上限略小于 maxconnectionsblocking无连接时是否等待生产建议 Truemaxusage单连接复用次数1000–5000配合 ping 使用ping取连接时探活4查询前检查maxidle空闲回收秒数300避免连接被 DB 侧先断封装一个 SQLHelper保证每次请求独立取连接、用完归还# utils/sql.py import pymysql from settings import PYMYSQL_POOL class SQLHelper: staticmethod def open(cursor_clspymysql.cursors.DictCursor): conn PYMYSQL_POOL.connection() cursor conn.cursor(cursorcursor_cls) return conn, cursor staticmethod def close(conn, cursor): conn.commit() cursor.close() conn.close() # 归还到池不是真正关闭 TCP classmethod def fetch_one(cls, sql, argsNone): conn, cursor cls.open() try: cursor.execute(sql, args) return cursor.fetchone() finally: cls.close(conn, cursor) classmethod def fetch_all(cls, sql, argsNone): conn, cursor cls.open() try: cursor.execute(sql, args) return cursor.fetchall() finally: cls.close(conn, cursor)关键点conn.close()在 PooledDB 里是归还连接不是断开。如果你用with上下文要确认封装层正确处理了归还逻辑否则连接会泄漏。4. 验证请求压测与日志确认连接复用和超时回收配置写完必须验证否则你不知道连接到底有没有复用、有没有泄漏。分三步先看池的实时状态再压测观察连接数最后查日志确认回收。第一步在 Flask 里加一个调试接口打印池的当前状态# app.py from flask import Flask, jsonify from settings import PYMYSQL_POOL from utils.sql import SQLHelper app Flask(__name__) app.route(/pool/status) def pool_status(): return jsonify({ cached: PYMYSQL_POOL._cached, # 当前空闲连接数 connections: PYMYSQL_POOL._connections, # 已创建的连接总数 maxconnections: PYMYSQL_POOL.maxconnections, }) app.route(/users) def users(): rows SQLHelper.fetch_all(SELECT id, name FROM users LIMIT 10) return jsonify(rows)启动服务后先访问/pool/status应该看到cached等于mincached5connections也是 5说明预热生效。第二步用wrk或ab压测/users接口wrk -t4 -c50 -d30s http://127.0.0.1:5000/users压测期间反复访问/pool/status观察connections是否稳定在maxconnections以内。如果connections持续增长且不回落说明有连接没归还大概率是异常路径没走finally。第三步在 MySQL 侧确认连接数SHOW STATUS LIKE Threads_connected; SHOW PROCESSLIST;压测结束后等待maxidle300 秒再查一次Threads_connected应该回落到接近mincached的水平。如果长时间不降检查maxidle是否生效以及是否有长事务占着连接。日志侧建议在SQLHelper.close里加一行 debug 日志记录归还动作import logging logger logging.getLogger(__name__) staticmethod def close(conn, cursor): conn.commit() cursor.close() conn.close() logger.debug(connection returned to pool)压测时把日志级别调到 DEBUG统计归还次数和取用次数是否匹配。取用 1000 次、归还 998 次那 2 次就是泄漏点。5. 本篇常见错排查连接泄漏、超时与池耗尽报错PooledDB 连接池已满或请求卡住不返回。先看maxconnections是不是设太小再看有没有连接没归还。最常见的是视图函数里cursor.execute抛异常后直接 return没走close。用try/finally包住或者统一走SQLHelper的封装方法。报错(2006, MySQL server has gone away)。连接空闲太久被 MySQL 侧wait_timeout断开了但池里还认为它可用。把ping设为 4每次查询前探活同时把maxidle设得比 MySQL 的wait_timeout小让池主动回收。报错Commands out of sync。同一个连接上还有未读完的结果集就发了新查询。检查是不是在fetchone之后没读完就复用了 cursor或者多个线程共用了同一个连接对象。连接池保证每个请求独立取连接不要在请求之间传递 conn。压测时Threads_connected打满 DB 上限。池的maxconnections乘以 worker 数量超过了 DB 的max_connections。比如 4 个 gunicorn worker每个池 20就是 80 个连接。要么调小单池上限要么减少 worker 数要么提升 DB 上限。连接归还了但connections不降。maxcached设得太大闲置连接一直留在池里。把maxcached调到合理值配合maxidle让空闲连接自然回收。AI 调用报 401 或超时。检查TAOTOKEN_API_KEY环境变量是否注入base_url是否写成https://taotoken.net/api不带 UTM。超时的话调大config.toml里的timeout或者在客户端单独设置。6. 接入与排障按场景选对入口连接池跑通之后AI 工具侧的接入按你的实际场景选入口。如果是在排查接口报错、确认 Key 和通道是否正常先去 API Keys 页面核对密钥状态再对照接入文档检查base_url和请求头https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite如果只是想快速验证模型通道能不能通直接在模型对话页面发一条消息比写代码快https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite如果你在做长期的编码任务或者 Agent 工作流需要稳定的调用配额和更长的上下文支持看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite最后提醒一个实操细节config.toml里的password和secret_key不要提交到 git用.gitignore排除生产环境通过环境变量覆盖。连接池参数先按模板跑压测后再根据Threads_connected和connections的曲线微调不要一上来就把maxconnections拉满。 
长河君 · 2026/9/28 6:57:33 · 阅读 1.2k 【10.0】Flask框架之数据库连接池：TaoToken统一Key接入与config.toml配置骨架
1. 从全局 conn 到连接池Flask 高并发下的真实痛点Flask 里写数据库最容易踩的坑就是把conn和cursor定义成模块级全局变量。单进程调试时看着没问题一旦上 gunicorn 多 worker 或者开多线程多个请求同时操作同一个连接对象轻则结果串台重则直接抛Packet sequence number wrong或者Commands out of sync。我试过在本地用threadedTrue跑一个查询接口20 个并发就把全局 cursor 打崩了。把连接挪进视图函数能解决串台但每次请求都pymysql.connect()再close()高并发下 TCP 握手和 MySQL 认证的开销非常可观。压测 500 QPS 时光是建连就吃掉大量 CPU数据库侧Threads_connected频繁抖动。连接池要解决的就是这两件事连接复用 并发隔离。这篇聚焦 Flask 应用在高并发下数据库连接池的配置与验证同时把 AI 工具侧的接入统一到 TaoToken 的 Key/API 通道上。你会拿到一份可复制的config.toml配置骨架、连接池参数模板以及通过压测和日志验证连接复用、超时回收、连接泄漏的具体动作。适合已经写过 Flask 接口、准备上生产或者正在排查连接数暴涨的开发者。2. TaoToken 前置统一 Key 与 API 通道准备在写连接池代码之前先把 AI 工具侧的接入通道理清楚。TaoToken 提供统一的 Key 和 API 入口Flask 项目里如果要用到模型对话、代码补全这类能力不需要每个工具单独配一套凭证统一走一个 Key 即可。你需要先拿到 API Key。打开控制台创建密钥https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite创建完成后在 API Keys 页面管理你的密钥建议按项目或环境分开建方便后续排查是哪个服务在调用https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteAPI 的基础地址是https://taotoken.net/api注意这个地址不带 UTM 参数直接用于代码里的base_url。如果你在 Flask 里封装 AI 调用客户端把 Key 放进环境变量不要硬编码进config.toml提交到仓库。注意config.toml里只放非敏感的连接池参数和 API 地址Key 一律走环境变量注入。接入文档在这里里面有各语言 SDK 的调用示例Flask 项目可以直接参考 Python 部分https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite如果你后续要做长期编码或者 Agent 类任务可以了解 Coding Plan它更适合持续性的代码生成场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite想先验证模型通道是否通可以直接在模型对话页面发一条测试消息https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite3. 可复制配置config.toml 骨架与连接池参数模板下面这份config.toml是完整的配置骨架包含数据库连接池参数和 AI 通道地址。用tomllibPython 3.11或者tomli读取Flask 启动时加载一次。# config.toml [app] debug false secret_key change-me-in-env [database] host 127.0.0.1 port 3306 user app_user password app_password database app_db charset utf8mb4 [database.pool] # 连接池允许的最大连接数0 和 None 表示不限制 maxconnections 20 # 初始化时至少创建的空闲连接数 mincached 5 # 池中最多闲置的连接数0 和 None 不限制 maxcached 10 # 池中最多共享的连接数pymysql threadsafety1 时该值实际不生效 maxshared 0 # 无可用连接时是否阻塞等待True 等待False 直接报错 blocking true # 单个连接最多被重复使用的次数None 表示无限制 maxusage 1000 # 每次取连接时 ping 服务端0never 1default 2when cursor created 4when query executed 7always ping 4 # 连接空闲回收时间秒超过后从池中移除 maxidle 300 [ai] base_url https://taotoken.net/api # api_key 从环境变量 TAOTOKEN_API_KEY 读取不写在这里 timeout 30读取配置并初始化连接池的代码# settings.py import os import tomllib import pymysql from dbutils.pooled_db import PooledDB with open(config.toml, rb) as f: CFG tomllib.load(f) DB CFG[database] POOL_CFG DB[pool] PYMYSQL_POOL PooledDB( creatorpymysql, maxconnectionsPOOL_CFG[maxconnections], mincachedPOOL_CFG[mincached], maxcachedPOOL_CFG[maxcached], maxsharedPOOL_CFG[maxshared], blockingPOOL_CFG[blocking], maxusagePOOL_CFG[maxusage], pingPOOL_CFG[ping], maxidlePOOL_CFG[maxidle], hostDB[host], portDB[port], userDB[user], passwordDB[password], databaseDB[database], charsetDB[charset], ) TAOTOKEN_API_KEY os.environ.get(TAOTOKEN_API_KEY, ) TAOTOKEN_BASE_URL CFG[ai][base_url]参数对照表方便你按实际负载调整参数作用高并发建议值maxconnections池上限超过则阻塞或报错按 DB 最大连接数的 60% 设置mincached启动时预热连接数5–10避免首波请求建连maxcached闲置连接上限略小于 maxconnectionsblocking无连接时是否等待生产建议 Truemaxusage单连接复用次数1000–5000配合 ping 使用ping取连接时探活4查询前检查maxidle空闲回收秒数300避免连接被 DB 侧先断封装一个 SQLHelper保证每次请求独立取连接、用完归还# utils/sql.py import pymysql from settings import PYMYSQL_POOL class SQLHelper: staticmethod def open(cursor_clspymysql.cursors.DictCursor): conn PYMYSQL_POOL.connection() cursor conn.cursor(cursorcursor_cls) return conn, cursor staticmethod def close(conn, cursor): conn.commit() cursor.close() conn.close() # 归还到池不是真正关闭 TCP classmethod def fetch_one(cls, sql, argsNone): conn, cursor cls.open() try: cursor.execute(sql, args) return cursor.fetchone() finally: cls.close(conn, cursor) classmethod def fetch_all(cls, sql, argsNone): conn, cursor cls.open() try: cursor.execute(sql, args) return cursor.fetchall() finally: cls.close(conn, cursor)关键点conn.close()在 PooledDB 里是归还连接不是断开。如果你用with上下文要确认封装层正确处理了归还逻辑否则连接会泄漏。4. 验证请求压测与日志确认连接复用和超时回收配置写完必须验证否则你不知道连接到底有没有复用、有没有泄漏。分三步先看池的实时状态再压测观察连接数最后查日志确认回收。第一步在 Flask 里加一个调试接口打印池的当前状态# app.py from flask import Flask, jsonify from settings import PYMYSQL_POOL from utils.sql import SQLHelper app Flask(__name__) app.route(/pool/status) def pool_status(): return jsonify({ cached: PYMYSQL_POOL._cached, # 当前空闲连接数 connections: PYMYSQL_POOL._connections, # 已创建的连接总数 maxconnections: PYMYSQL_POOL.maxconnections, }) app.route(/users) def users(): rows SQLHelper.fetch_all(SELECT id, name FROM users LIMIT 10) return jsonify(rows)启动服务后先访问/pool/status应该看到cached等于mincached5connections也是 5说明预热生效。第二步用wrk或ab压测/users接口wrk -t4 -c50 -d30s http://127.0.0.1:5000/users压测期间反复访问/pool/status观察connections是否稳定在maxconnections以内。如果connections持续增长且不回落说明有连接没归还大概率是异常路径没走finally。第三步在 MySQL 侧确认连接数SHOW STATUS LIKE Threads_connected; SHOW PROCESSLIST;压测结束后等待maxidle300 秒再查一次Threads_connected应该回落到接近mincached的水平。如果长时间不降检查maxidle是否生效以及是否有长事务占着连接。日志侧建议在SQLHelper.close里加一行 debug 日志记录归还动作import logging logger logging.getLogger(__name__) staticmethod def close(conn, cursor): conn.commit() cursor.close() conn.close() logger.debug(connection returned to pool)压测时把日志级别调到 DEBUG统计归还次数和取用次数是否匹配。取用 1000 次、归还 998 次那 2 次就是泄漏点。5. 本篇常见错排查连接泄漏、超时与池耗尽报错PooledDB 连接池已满或请求卡住不返回。先看maxconnections是不是设太小再看有没有连接没归还。最常见的是视图函数里cursor.execute抛异常后直接 return没走close。用try/finally包住或者统一走SQLHelper的封装方法。报错(2006, MySQL server has gone away)。连接空闲太久被 MySQL 侧wait_timeout断开了但池里还认为它可用。把ping设为 4每次查询前探活同时把maxidle设得比 MySQL 的wait_timeout小让池主动回收。报错Commands out of sync。同一个连接上还有未读完的结果集就发了新查询。检查是不是在fetchone之后没读完就复用了 cursor或者多个线程共用了同一个连接对象。连接池保证每个请求独立取连接不要在请求之间传递 conn。压测时Threads_connected打满 DB 上限。池的maxconnections乘以 worker 数量超过了 DB 的max_connections。比如 4 个 gunicorn worker每个池 20就是 80 个连接。要么调小单池上限要么减少 worker 数要么提升 DB 上限。连接归还了但connections不降。maxcached设得太大闲置连接一直留在池里。把maxcached调到合理值配合maxidle让空闲连接自然回收。AI 调用报 401 或超时。检查TAOTOKEN_API_KEY环境变量是否注入base_url是否写成https://taotoken.net/api不带 UTM。超时的话调大config.toml里的timeout或者在客户端单独设置。6. 接入与排障按场景选对入口连接池跑通之后AI 工具侧的接入按你的实际场景选入口。如果是在排查接口报错、确认 Key 和通道是否正常先去 API Keys 页面核对密钥状态再对照接入文档检查base_url和请求头https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite如果只是想快速验证模型通道能不能通直接在模型对话页面发一条消息比写代码快https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite如果你在做长期的编码任务或者 Agent 工作流需要稳定的调用配额和更长的上下文支持看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite最后提醒一个实操细节config.toml里的password和secret_key不要提交到 git用.gitignore排除生产环境通过环境变量覆盖。连接池参数先按模板跑压测后再根据Threads_connected和connections的曲线微调不要一上来就把maxconnections拉满。