1. 从连接泄漏说起Python 多线程写库为什么会把 MySQL 拖垮如果你用 Python 直接开多线程操作 MySQL大概率会遇到两个现象一是程序跑着跑着报Too many connections二是线程之间抢同一个 cursor数据错乱或者直接抛异常。很多人第一反应是给 cursor 加锁threading.Lock一上结果发现速度比单线程还慢——因为锁把并发彻底串行化了加锁这个行为本身就是反其道而行。问题的根子不在锁而在连接。MySQL 服务端每个连接都要占一个线程和一块内存默认max_connections也就 151。你开 50 个 Python 线程每个线程pymysql.connect()一次再忘记close()连接数只增不减几分钟就把数据库连接槽吃满。这时候新请求全部卡在Cant connect to MySQL server上整个服务雪崩。连接池要解决的就是这件事预先维护一批连接谁用谁取用完还回去而不是每次新建再销毁。dbutils里的PooledDB是 Python 生态里最轻量、最不挑框架的方案不依赖 Django/Flask纯脚本、定时任务、爬虫都能直接用。它内部维护一个空闲连接队列connection()时从队列拿close()时不是真关而是归还。这样连接数被硬性限制在maxconnections以内天然收敛。我试过在一个日更百万行的数据清洗脚本里把裸pymysql换成PooledDB同样的 8 线程连接数从峰值 200 稳定压到 12Threads_connected曲线从锯齿变成一条平线。下面这套路径就是我当时踩完坑整理出来的先搭池再调参最后用压测确认连接数真的收敛。适合谁看正在用 Python 多线程/多进程写 MySQL 的开发者被连接泄漏折磨过的运维以及想把脚本从「能跑」升级到「跑得稳」的人。核心检索词就三个Python 数据库连接池、PooledDB、连接泄漏排查。2. 前置准备装 dbutils、确认驱动、把配置从代码里抽出来动手之前先把依赖和配置这两件事理清楚后面调参会顺很多。安装dbutils和驱动。PooledDB本身不带数据库驱动它只是个池管理器真正连库的还是pymysqlMySQL、pymssqlSQL Server、cx_OracleOracle这些。所以两个都要装pip install dbutils pymysql如果你用mysqlclient或PyMySQL都行creator参数传哪个模块就用哪个。我习惯pymysql纯 Python装起来不折腾编译。把连接信息从代码里抽出来。excerpt 里那段代码把 host、password 硬编码在__init__里本地调试无所谓一上生产就是隐患。用configparser读setting.ini改配置不用动代码[USER] MYSQL_HOST 192.168.30.246 MYSQL_PORT 3306 MYSQL_USER root MYSQL_PASSWORD 123456 MYSQL_DB bpq读取时注意.strip( )把引号和空格去掉否则int()转换会炸。这一步看着琐碎但连接池参数调优时你会频繁改maxconnections、blocking配置外置能省掉大量重启改代码的时间。确认数据库的max_connections。登录 MySQL 执行SHOW VARIABLES LIKE max_connections; SHOW STATUS LIKE Threads_connected;前者是服务端上限后者是当前连接数。你的池maxconnections乘以进程数不能超过服务端上限否则池内还没满服务端先拒了。比如max_connections151你开 4 个进程每个池最多给 30留点余量给运维连接。还有一个容易忽略的点ping参数。MySQL 默认wait_timeout是 8 小时连接空闲太久会被服务端单方面掐断池里拿到的就是死连接。ping4表示执行 SQL 前 ping 一次发现断了自动重连。这个参数是连接池稳定复用的关键后面调参章节会细说。3. 可复制配置PooledDB 初始化参数逐项拆解与完整代码这一节是全文核心我把PooledDB的初始化参数按「必调」和「可默认」分开讲然后给一份可以直接抄的完整类。先看参数对照表这是调优的地图参数默认值作用生产建议creator无数据库驱动模块必填如 pymysqlmincached0初始化时创建的空闲连接数保持 0避免库不可用时项目起不来maxcached0池中空闲连接上限设成 maxconnections 同量级如 200maxshared0共享连接上限保持 0每连接专用maxconnections0允许的最大连接数按服务端上限/进程数算如 30blockingFalse池满时新请求是否阻塞必设 True否则池满直接报错maxusage0单连接最大复用次数保持 0无限制setsession无会话准备 SQL 列表需要设时区时用resetTrue归还时是否回滚保持 Trueping1何时 ping 检查连接设 4执行 SQL 前检查重点说三个。blockingTrue是必须的默认False意味着连接数达到maxconnections时connection()直接抛异常高并发下这就是定时炸弹设成True后新请求会阻塞等待等别人归还再拿行为可预期。ping4解决死连接问题代价是每次执行前多一次轻量往返换来的是不会拿到被服务端掐断的连接。maxcached控制空闲连接不无限增长池用完后多余连接会被真正关闭避免占着服务端连接不放。下面是完整可复制代码把配置读取、池初始化、取连接、归还、查询、更新都封装好import configparser import threading import time import pymysql from dbutils.pooled_db import PooledDB class MysqlPool: def __init__(self, config_pathsetting.ini): self.config configparser.RawConfigParser() self.config.read(config_path, encodingutf-8) self.host self.config.get(USER, MYSQL_HOST).strip( ) self.port int(self.config.get(USER, MYSQL_PORT).strip( )) self.user self.config.get(USER, MYSQL_USER).strip( ) self.password self.config.get(USER, MYSQL_PASSWORD).strip( ) self.db self.config.get(USER, MYSQL_DB).strip( ) self.pool None self._init_pool() def _init_pool(self): while True: try: self.pool PooledDB( creatorpymysql, mincached0, maxcached200, maxshared0, maxconnections30, blockingTrue, maxusage0, ping4, hostself.host, portself.port, userself.user, passwordself.password, dbself.db, charsetutf8mb4, ) print(数据库连接池初始化成功) break except BaseException as e: print(f连接池初始化失败: {e}5 秒后重试) time.sleep(5) def get_conn(self): conn self.pool.connection() cur conn.cursor() return conn, cur def close_conn(self, conn, cur): cur.close() conn.close() def query(self, sql, argsNone): conn, cur self.get_conn() try: cur.execute(sql, args) return cur.fetchall() finally: self.close_conn(conn, cur) def execute(self, sql, argsNone): conn, cur self.get_conn() try: cur.execute(sql, args) conn.commit() return cur.rowcount except BaseException as e: conn.rollback() raise e finally: self.close_conn(conn, cur)注意charset我改成了utf8mb4比utf8更完整能存 emoji 和四字节字符。query和execute里用try/finally保证连接一定归还这是防泄漏的第一道闸。execute里加了rollback出错时把事务回滚掉再归还避免脏事务污染下一个使用者。如果你用 Cline MCP 或 Codex 这类工具管理数据库连接配置记得三件套要写全Base URL、Key、Model ID 各就各位别只填一半。不过PooledDB本身是纯 Python 库不依赖这些配置写进setting.ini就够了。4. 验证请求压测看连接数是否真的收敛池搭好了不验证等于没搭。这一节给你一套可执行的压测动作用数据确认连接数收敛。先写一个多线程压测脚本模拟 20 个线程各跑 200 次查询import threading import time from mysql_pool import MysqlPool pool MysqlPool() results [] lock threading.Lock() def worker(tid): for i in range(200): rows pool.query(SELECT id FROM bmms_image LIMIT 1) with lock: results.append((tid, i, len(rows))) threads [threading.Thread(targetworker, args(t,)) for t in range(20)] start time.time() for t in threads: t.start() for t in threads: t.join() print(f总耗时 {time.time() - start:.2f}s完成 {len(results)} 次查询)跑的同时在 MySQL 里盯连接数SHOW STATUS LIKE Threads_connected; SHOW STATUS LIKE Max_used_connections;预期结果Threads_connected稳定在maxconnections30以内不会随线程数线性上涨。Max_used_connections记录历史峰值如果它等于 30说明池被用满过blockingTrue起了作用请求在排队而不是报错。如果Threads_connected一路涨到几百说明有连接没归还回到第 5 节排查。再验证一下连接复用。在池初始化后打印一次连接 id压测中再打印看是否复用conn, cur pool.get_conn() cur.execute(SELECT CONNECTION_ID()) print(连接 ID:, cur.fetchone()) pool.close_conn(conn, cur)同一个连接被反复取用时CONNECTION_ID()应该重复出现。如果每次都是新 id说明maxcached太小或者连接被提前关了。压测通过的标准就两条连接数收敛在maxconnections以内且没有Too many connections报错。达到这两条池就算用稳了。5. 常见报错排查401、local proxy failed、reading choices、OAuth 对照这一节把实际会撞到的报错列出来对照着查。pymysql.err.OperationalError: (1040, Too many connections)。这是最典型的泄漏信号。原因通常是close_conn没在finally里调用或者异常路径跳过了归还。检查你的每个get_conn是否都有配对的close_conn且放在finally。另一个可能是maxconnections设太大超过服务端max_connections把池调小。pymysql.err.InterfaceError: (0, )或Lost connection to MySQL server during query。这是死连接空闲太久被服务端掐了。把ping设成 4让每次执行前检查。如果还报检查 MySQL 的wait_timeout必要时调大或让池定期保活。local proxy failed这类报错通常出现在你通过某个本地代理层连数据库时代理进程挂了或端口没监听。先确认代理服务在跑再确认host/port指向的是代理而不是直连地址。这类问题和连接池无关是链路问题。reading choices报错一般出现在配置解析阶段比如configparser读到的值带了引号导致int()失败或者setting.ini编码不对。用encodingutf-8打开.strip( )去引号基本能解决。OAuth相关报错和PooledDB无关那是认证层的事。如果你在用某些云数据库的 OAuth 鉴权连接串里要带 tokentoken 过期会报鉴权失败。这种场景下ping重连也救不了需要刷新 token 后重建池。还有一个隐蔽的坑blockingTrue时池满会阻塞如果某个线程拿了连接不还其他线程会一直等表现为程序卡死而不是报错。用SHOW PROCESSLIST看有没有长时间Sleep的连接配合代码里的finally排查。排查顺序建议先看Threads_connected是否超maxconnections再看报错类型最后回到代码检查finally。大部分泄漏都是finally漏写导致的。6. 把连接池用稳之后参数微调与长期维护建议池跑起来只是开始长期稳定靠的是参数随业务微调。maxconnections不是越大越好它受服务端max_connections和进程数双重约束。单进程脚本给 20 到 30 够用Web 服务按 worker 数分摊。maxcached设成和maxconnections同量级让空闲连接有地方待又不至于占着服务端资源。ping4有轻微性能开销如果你的库连接很稳定、wait_timeout调得很大可以降到ping1或ping2。但生产环境我建议保持 4稳定性优先。监控上定期采集Threads_connected和Max_used_connections画成曲线。正常应该是平线一旦出现持续上涨就是泄漏前兆趁早查。把这两个指标接进你的告警比出事后再救火省心得多。最后连接池的配置和代码建议纳入版本管理setting.ini里的密码用环境变量注入别提交到仓库。池初始化那段while True重试逻辑保留数据库短暂不可用时能自愈但重试间隔别太短5 秒是个合理值。需要对照官方参数说明或拿接入凭证时可以走这几个入口模型对话在 https://taotoken.net/api 对应的对话页API Keys 在 console 的 api-keys 页接入文档在 doc 页长期编码和 Agent 场景看 coding-plan。把 Base URL、Key、Model ID 三件套配齐再回到本文的池配置整条链路就通了。