很多后端开发刚开始用 Python 碰 MySQL最常踩的坑就是“代码能跑一上生产就崩”。连接超时、数据对不上、并发一高数据库直接卡死这些问题十有八九不是 SQL 写错了而是对连接管理、事务边界和操作方式的理解还停留在“能用就行”。这篇文章我不打算写那种面面俱到的 API 手册而是把实际项目里真正会用到的核心能力拆开揉碎从最基础的增删改查到事务隔离级别怎么选再到连接池参数怎么调最后聊几个压测时才能发现的性能优化细节。无论你是刚入门 Python 想写通第一个 MySQL 程序还是已经写过一些业务代码但总感觉哪里不对劲这篇文章应该都能帮你把这块短板补上。1. 环境准备与驱动选型别在第一步就埋雷1.1 PyMySQL 与 MySQL Connector/Python 怎么选Python 操作 MySQL 的官方驱动叫 mysql-connector-python由 Oracle 维护支持的特性最全但说实话实际项目里用 PyMySQL 的人反而更多。原因很简单PyMySQL 是纯 Python 实现安装没有任何编译依赖装完就能用而且接口风格跟 MySQLdb 高度一致很多老项目的迁移成本极低。选型这件事我建议这样看如果你的项目跑在 Linux 生产环境而且对性能有极致要求可以考虑用 mysqlclient它是 MySQLdb 的分支基于 C 扩展实现速度确实快但安装依赖 libmysqlclient-dev编译出问题的情况不少。如果你只是想快速上手、写业务逻辑、不折腾环境PyMySQL 是最稳的选择。Connector/Python 我一般只在需要最新 MySQL 特性的时候才用日常开发很少碰。安装很简单直接 pip 装就行。pip install pymysql装完可以用一段代码验证环境是否正常这一步能筛掉 80% 后面会遇到的问题。import pymysql conn pymysql.connect( host127.0.0.1, port3306, userroot, passwordyour_password, charsetutf8mb4 ) print(conn.ping()) conn.close()1.2 建库建表字符集和存储引擎一次说清很多新手在建表的时候不指定字符集默认落到 latin1后面写入中文直接变乱码排查起来特别浪费时间。我常用的建库建表语句如下。CREATE DATABASE IF NOT EXISTS demo DEFAULT CHARACTER SET utf8mb4 DEFAULT COLLATE utf8mb4_general_ci; USE demo; CREATE TABLE IF NOT EXISTS user ( id INT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 主键, username VARCHAR(64) NOT NULL COMMENT 用户名, email VARCHAR(128) NOT NULL DEFAULT COMMENT 邮箱, age INT UNSIGNED NOT NULL DEFAULT 0 COMMENT 年龄, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 创建时间, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 更新时间, PRIMARY KEY (id), UNIQUE KEY uk_username (username) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_general_ci COMMENT用户表;这里面有几个关键点我重点说一下。用 utf8mb4 而不是 utf8是因为 MySQL 的 utf8 最多只支持 3 个字节像 emoji 这类 4 字节字符根本存不进去。InnoDB 是必须的它是目前唯一支持事务和外键的存储引擎MyISAM 现在除了极少数只读场景基本可以放弃。id 用 INT UNSIGNED如果预估数据量会超过 40 亿就直接上 BIGINT不然后期改表结构非常痛苦。建表的逻辑我再多啰嗦一句id 和 created_at 这种字段尽量让数据库自己生成不要让应用层传值。这样能避免很多分布式场景下的 ID 冲突也方便后续做数据归档和分库分表。2. 增删改查核心操作从能跑到写对2.1 连接数据库游标 (Cursor) 到底是什么Python 操作 MySQL 的基本流程是固定的四步建立连接、创建游标、执行 SQL、关闭连接。游标这个概念很多人一开始理解不了我打个比方连接就像一根管子接到数据库上游标就是你拿在手里控制数据流动的那根手柄。所有查询结果都要通过游标来获取。基本的连接代码结构如下。import pymysql conn pymysql.connect( host127.0.0.1, port3306, userroot, passwordyour_password, databasedemo, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor ) try: with conn.cursor() as cursor: cursor.execute(SELECT * FROM user WHERE id %s, (1,)) row cursor.fetchone() print(row) conn.commit() finally: conn.close()这里有两个细节我要特别提醒。第一个是 cursorclass 设置成 DictCursor这样查出来的每一行是字典字段名可以直接当 key 用比默认的元组好读得多。第二个是参数占位符用 %s不要自己拼 SQL。后面我会专门展开讲这个这是防注入的关键。2.2 INSERT、UPDATE、DELETE别漏掉 commit增删改操作在 PyMySQL 里的套路完全一致execute 之后必须 commit不 commit 数据不会真正落库。这是新手最容易踩的坑代码执行没报错数据就是查不到原因就在这。先看插入的写法。with conn.cursor() as cursor: sql INSERT INTO user (username, email, age) VALUES (%s, %s, %s) affected cursor.execute(sql, (zhangsan, zhangsanexample.com, 25)) print(f影响行数: {affected}) print(f自增ID: {cursor.lastrowid}) conn.commit()cursor.lastrowid 能拿到刚插入记录的自增 ID这在后续处理主外键关联时特别有用省一次多余的查询。批量插入是另一个高频需求写法跟单条插入的区别就在 executemany。data [ (user1, user1example.com, 20), (user2, user2example.com, 21), (user3, user3example.com, 22), ] with conn.cursor() as cursor: sql INSERT INTO user (username, email, age) VALUES (%s, %s, %s) affected cursor.executemany(sql, data) print(f批量插入行数: {affected}) conn.commit()执行批量插入时要注意单次数据量一次塞一万条以上容易出现 packet 过大错误常见的做法是每 1000 条左右作为一批提交。更新和删除逻辑类似。# 更新 with conn.cursor() as cursor: sql UPDATE user SET age %s WHERE username %s affected cursor.execute(sql, (26, zhangsan)) print(f更新行数: {affected}) conn.commit() # 删除 with conn.cursor() as cursor: sql DELETE FROM user WHERE username %s affected cursor.execute(sql, (user3,)) print(f删除行数: {affected}) conn.commit()我想在这里多说一句UPDATE 和 DELETE 没有 WHERE 条件就是全表操作。代码里只要出现这种 SQL一定要养成先 SELECT 查一下影响范围再执行的坏习惯克星或者直接在事务里操作发现不对马上回滚。2.3 SELECT 查询fetchone、fetchmany、fetchall 的取舍查询结果的读取有三种方式使用场景完全不同。fetchone()读一行适合按主键查详情。fetchmany(size)读指定行数适合分页加载一批处理一批。fetchall()一次读出全部结果适合小数据量展示。# 单条查询 with conn.cursor() as cursor: cursor.execute(SELECT * FROM user WHERE id %s, (1,)) user cursor.fetchone() print(user) # 批量查询每批100条 with conn.cursor() as cursor: cursor.execute(SELECT * FROM user WHERE age %s, (18,)) while True: batch cursor.fetchmany(100) if not batch: break for row in batch: print(row)fetchall 千万要慎用。我刚工作那会儿处理一张千万级的表直接 fetchall内存瞬间被打满进程直接 OOM。如果只是从头到尾遍历用 fetchmany 或者服务端游标才是正确姿势。2.4 参数化查询防 SQL 注入的底线SQL 注入的原理我不多解释只强调一点永远不要用字符串拼接的方式把变量塞进 SQL。正确的做法是用参数化查询让驱动帮你处理转义。# 错误示范千万不要这么写 username zhangsan OR 11 sql fSELECT * FROM user WHERE username {username} cursor.execute(sql) # 这条能查出全表数据 # 正确写法 sql SELECT * FROM user WHERE username %s cursor.execute(sql, (username,))有人觉得参数化查询麻烦觉得反正项目没那么容易被注入这种想法非常危险。数据库里一旦出问题轻则数据泄露重则整库被删。养成习惯所有动态参数一律走 %s 占位没有任何例外。3. 事务控制数据一致性的最后防线3.1 事务四大特性 ACID 到底说的是什么很多开发提到事务就说“要么全成功要么全失败”这话只说对了一半。事务的完整定义包括四个方面原子性 (Atomicity)、一致性 (Consistency)、隔离性 (Isolation) 和持久性 (Durability)。拿转账场景来理解。A 给 B 转 100 块从 A 账户扣 100 和往 B 账户加 100 是一个原子操作这说的是原子性。事务完成后所有账户余额总和不变这是一致性。两个事务同时操作同一笔钱互相之间不能产生错误干扰这是隔离性。事务一旦提交数据就不允许丢重启数据库也一样这是持久性。InnoDB 通过 redo log 保证持久性和原子性通过 undo log 保证回滚能力通过锁和 MVCC 实现隔离性。Python 代码层面需要做的就是明确事务边界该提交提交该回滚回滚。3.2 Python 中实现事务commit 与 rollback 的正确姿势PyMySQL 默认开启事务执行 DML 语句后必须手动 commit。Python 里管理事务最优雅的方式是用 try / except / else 结构。try: with conn.cursor() as cursor: cursor.execute(UPDATE account SET balance balance - 100 WHERE user_id %s, (1,)) cursor.execute(UPDATE account SET balance balance 100 WHERE user_id %s, (2,)) conn.commit() except Exception as e: conn.rollback() print(f执行失败已回滚: {e}) finally: conn.close()这个结构保证要么两条 UPDATE 全部生效要么全部不生效。最常见的错误是第一个 UPDATE 执行成功后第二个报错代码里没有 except直接往外抛异常结果第一条数据已经改了第二条没改上数据就错了。我另外补充一个细节用 with conn.cursor() 管理游标是自动关闭游标但连接上的事务不会自动提交或回滚。所以 with 块内更保险的做法是最后显式调用 commit异常时在 except 中 rollback。3.3 事务隔离级别脏读、不可重复读与幻读MySQL 默认的事务隔离级别是 REPEATABLE READ可重复读具体有四种级别按隔离强度从低到高排列如下。隔离级别脏读不可重复读幻读说明READ UNCOMMITTED可能可能可能基本不用READ COMMITTED不会可能可能Oracle 默认REPEATABLE READ不会不会可能InnoDB 实际解决了MySQL 默认SERIALIZABLE不会不会不会性能最差这几种异常的通俗解释是这样的。脏读就是事务 A 读到事务 B 还没提交的数据结果 B 回滚了A 读到的就是无效数据。不可重复读事务 A 里同一查询执行两次结果不一样因为其他事务在这期间提交了修改。幻读事务 A 按条件查出来一批行期间事务 B 插入了新行A 再查多出来了几条。InnoDB 在 REPEATABLE READ 级别下通过 MVCC 和间隙锁基本解决了幻读问题这也是它能成为默认级别的原因。Python 层面设置隔离级别的方法如下。conn pymysql.connect( host127.0.0.1, userroot, passwordyour_password, databasedemo, charsetutf8mb4 ) with conn.cursor() as cursor: cursor.execute(SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED) try: with conn.cursor() as cursor: cursor.execute(UPDATE account SET balance balance - 100 WHERE user_id %s, (1,)) conn.commit() except Exception: conn.rollback() finally: conn.close()事务级别并非越高越好。SERIALIZABLE 能杜绝所有并发问题代价是性能断崖式下跌绝大多数业务根本用不上。具体怎么选要结合业务容忍度来分析。4. 连接池设计高并发下数据库不被打垮的关键4.1 每次请求新建连接到底错在哪里最朴素的数据库操作方式是每个请求来的时候建立连接处理完关闭。这种做法在并发量低的时候没问题一上量就崩。原因有三个。第一每次建连都要经过 TCP 三次握手、MySQL 权限验证、连接初始化这个过程毫秒级起步高并发下累加起来非常可观。第二MySQL 服务端对连接数有限制默认 max_connections 一般是 151连接一多直接报 Too many connections。第三频繁建连和断连给数据库带来大量额外负载会让整体响应时间明显变长。连接池的思路特别像银行网点。如果每个人办业务都重新建一个柜台银行大厅早就挤爆了。连接池就是预先开好一批柜台来的人排队处理处理完柜台不撤下一批接着用。4.2 使用 dbutils 实现连接池参数配置与踩坑记录Python 生态里最常用的连接池工具是 DBUtils 的 PooledDB。安装方式如下。pip install DBUtils基础配置如下。import pymysql from dbutils.pooled_db import PooledDB pool PooledDB( creatorpymysql, maxconnections20, mincached5, maxcached10, maxshared10, blockingTrue, maxusageNone, setsession[], ping1, host127.0.0.1, port3306, userroot, passwordyour_password, databasedemo, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor ) conn pool.connection() try: with conn.cursor() as cursor: cursor.execute(SELECT 1) print(cursor.fetchone()) finally: conn.close()关键参数我逐个说。maxconnections 是连接池允许的最大连接数超过这个数就得排队等待设太小并发稍高就阻塞设太大数据库会被打垮。我一般按应用实例数乘以单个实例预估最大并发来定。mincached 是初始化时就建好的空闲连接数避免刚启动时请求来了现建连。maxcached 是空闲连接的最大缓存数超过这个数量的空闲连接会被关闭释放。blocking 设为 True连接耗尽时新请求阻塞等待而不是直接报错。ping1 表示取连接时如果连接空闲超过一定时间就发送 ping 包检查连接是否有效能避免拿到数据库已断开但应用不知道的死连接。有个坑我要重点提示从连接池拿连接用完一定记得 close但这个 close 不是真的关闭连接而是归还给池子。如果忘了这一步连接会被一直占用最后池子耗尽整个应用卡死。4.3 生产环境连接池方案对比与选型DBUtils 的 PooledDB 够用但它的连接池是单进程内的。如果应用是多进程部署每个进程都有自己的池子总连接数是进程数乘池大小规划时要把这个乘数算进去。实际生产里我见过更主流的方案是用 SQLAlchemy 作为 ORM 层它内部自带连接池而且对 PyMySQL 和 mysqlclient 做了统一封装切驱动只改一个 URL 就行。from sqlalchemy import create_engine engine create_engine( mysqlpymysql://root:your_password127.0.0.1:3306/demo?charsetutf8mb4, pool_size10, max_overflow5, pool_pre_pingTrue, pool_recycle3600 )SQLAlchemy 的 pool_size 相当于基础连接数max_overflow 是池满之后最多还能额外创建的连接数pool_recycle 是连接最大存活时间到点强制回收重建能有效规避 MySQL 的 wait_timeout 导致的连接失效问题。这组参数我实测下来稳定性很好推荐直接抄。如果并发规模到了单库扛不住的程度那就不是 Python 层能解决的问题了需要上 Proxy 层或者中间件做读写分离和分库分表比如 ShardingSphere、MyCat、ProxySQL 这类组件但它们都属于另一个领域这里不展开。5. 性能优化实战从索引到批量写入的全面提速5.1 索引优化原理回表、覆盖索引与最左前缀MySQL 加快查询速度最核心的手段是索引。InnoDB 的索引结构是 B 树主键索引的叶子节点存整行数据二级索引的叶子节点存主键值。用二级索引查询时先找到主键再回到主键索引里找完整行这个过程叫回表。回表是有成本的所以就有了覆盖索引的概念。如果查询需要的字段全部在二级索引里就不需要回表。这就是为什么尽量别用 SELECT *只查需要的字段配合合适的联合索引能大幅降低 IO。联合索引要遵守最左前缀原则。比如建一个 (username, age) 的联合索引实际上相当于建了 (username) 和 (username, age) 两个索引。如果直接拿 age 查这个索引是用不上的。索引建多了更新慢建少了查询慢这是典型的空间换时间取舍。生产环境一般先用 EXPLAIN 看执行计划确认走了哪个索引、扫了多少行再决定要不要加索引或者调整索引顺序。EXPLAIN SELECT * FROM user WHERE username zhangsan\G看 type 字段从好到差依次是 system、const、eq_ref、ref、range、index、ALL。ALL 表示全表扫描这种语句出现在高频查询里就是性能隐患必须处理。5.2 批量写入效率对比execute 和 executemany 差多少我曾经在一个项目里导入 50 万条历史数据一开始逐条 execute跑了二十多分钟还是没跑完后来改成 executemany 每批 1000 条几十秒就完成了。差距源于网络 IO。逐条插入相当于每条数据一个往返批量插入一次往返塞几千条。批量插入的写法在第 2 节已经展示过这里补充一个数据分批处理的完整示例。import pymysql conn pymysql.connect(host127.0.0.1, userroot, passwordyour_password, databasedemo, charsetutf8mb4) cursor conn.cursor() rows [(fuser_{i}, fuser_{i}example.com, i % 50) for i in range(500000)] batch_size 1000 for start in range(0, len(rows), batch_size): batch rows[start:start batch_size] cursor.executemany( INSERT INTO user (username, email, age) VALUES (%s, %s, %s), batch ) conn.commit() print(f已插入 {start len(batch)} 条) cursor.close() conn.close()这里 commit 的时机值得讨论一下。每批 commit 一次中途出错了最多丢一批可以重新跑。如果全部插完再 commit要么全成要么全失败大批量任务中途失败重来的成本很高。到底怎么选看你业务对一致性的容忍度。对于几百万行级别的大批量导入Python 循环 executemany 仍然偏慢更快的方案是先把数据写到 CSV 文件然后用 MySQL 的 LOAD DATA INFILE 导入那个速度差不多能再快一个数量级。这也是我强烈建议掌握的技巧。5.3 慢查询日志定位与 Python 层优化如果页面越跑越慢先别急着改代码打开 MySQL 的慢查询日志看看到底是哪些 SQL 拖慢了整体性能。SET GLOBAL slow_query_log ON; SET GLOBAL long_query_time 1;超过 1 秒的 SQL 会被记录到日志文件里定位到具体语句之后再用 EXPLAIN 分析执行计划。绝大多数慢查询的根因就是缺索引、扫全表或者干脆忘写 WHERE。Python 层还有一些容易被忽略的优化点。连接参数里加 charset 和 cursorclass 都算基础操作还有两个参数建议加上。conn pymysql.connect( host127.0.0.1, userroot, passwordyour_password, databasedemo, charsetutf8mb4, read_timeout10, write_timeout10, autocommitFalse )read_timeout 和 write_timeout 防止数据库假死时请求一直挂着不返回。这个超时配置对于线上稳定性非常关键没有超时的连接在数据库出问题时会把应用线程全部占满直接雪崩。5.4 防误操作UPDATE / DELETE 之前先 SELECT最后分享一个工作习惯不算技术但真的能救命。执行高危 UPDATE 或者 DELETE 之前先把 WHERE 条件抄到 SELECT 里查一遍。-- 先确认影响范围 SELECT * FROM user WHERE age 60; -- 确认无误再执行更新 UPDATE user SET status 1 WHERE age 60;很多线上事故都源于手一抖多打一个条件或者少打一个条件影响了几百万人。这个习惯救过我很多次建议直接刻进肌肉记忆。另外一个要点是不要用 DELETE 清空大表正确做法是 TRUNCATE TABLE它不走事务、不记逐行日志、速度极快。而逻辑上需要保留表结构又要快速清数据的时候直接用 DROP TABLE 再重建也比 DELETE 快得多。DELETE 会把每一行的删除操作写进 binlog量大时对主从同步的影响很大。6. 常见问题与排查技巧实录6.1 问题速查表我整理了一下实际项目里出现频率最高的问题和解决方案。现象可能原因解决方案Access denied for user密码错误或权限不足检查用户权限和密码Unknown database数据库不存在确认数据库名正确Table doesnt exist表名写错或库选错检查表名和当前 databaseLost connection during query单条 SQL 执行时间超过 wait_timeout优化 SQL 或调大超时时间Packet too large批量插入单批次数据量过大减小批量大小设置 max_allowed_packetToo many connections连接数超过 max_connections使用连接池限制应用并发连接数Data too long for column插入数据超过字段定义长度检查字段长度或调整字段定义Duplicate entry for key违反唯一索引约束用 INSERT IGNORE 或 ON DUPLICATE KEY UPDATE6.2 MySQL 8.0 认证插件导致连接失败的特殊情况MySQL 8.0 默认的认证插件是 caching_sha2_password而 PyMySQL 老版本可能不支持连接时直接报 Authentication plugin caching_sha2_password cannot be loaded。遇到这个问题有两种解法。首选升级 PyMySQL 到最新版。pip install --upgrade pymysql如果升级后仍有问题就在 MySQL 里把用户认证方式改回 mysql_native_password。ALTER USER rootlocalhost IDENTIFIED WITH mysql_native_password BY your_password; FLUSH PRIVILEGES;我推荐优先升级驱动不要为了兼容去改数据库的认证方式那属于拉低安全水位来迁就老代码。6.3 排查思路从报错信息定位问题我排查 Python MySQL 问题的方法论就三步。第一步仔细读报错第一行绝大部分问题在异常信息里就已经写明白了。第二步如果是连接层面的问题用命令行 mysql -h host -P port -u user -p 手动连一下看是不是网络、账号、防火墙方面的问题。第三步如果是 SQL 层面的问题把 SQL 单独在数据库客户端跑一遍看执行计划和实际结果。这套流程看着简单能解决 90% 的问题。很多人在代码里各种瞎试不如回到根上用最小化场景还原问题。比如批量插入报错就先拿一条数据试试确认单条能过再排查批次的问题效率最高。我个人在实际操作中还有一个体会就是连接池的参数永远不要照抄别人的。连接数上限设多少取决于数据库配置、应用实例数、业务并发模型是要压测加监控调出来的不是拍脑袋定出来的。先给足配置跑一段时间看监控数据再逐步调整比一次到位靠谱得多。另外MySQL 服务端的 wait_timeout 默认 8 小时连接池里的连接如果长期空闲被服务端断开应用不知道还继续用就会出现连接似乎成功但一查询就报错的情况连接池的 ping 参数和 SQLAlchemy 的 pool_pre_ping 就是专门解决这个问题的一定不要省。