Python操作MySQL全攻略：从连接管理到性能优化

很多后端开发刚开始用 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( host="127.0.0.1", port=3306, user="root", password="your_password", charset="utf8mb4" ) 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) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_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( host="127.0.0.1", port=3306, user="root", password="your_password", database="demo", charset="utf8mb4", cursorclass=pymysql.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", "zhangsan@example.com", 25)) print(f"影响行数: {affected}") print(f"自增ID: {cursor.lastrowid}") conn.commit()

cursor.lastrowid 能拿到刚插入记录的自增 ID，这在后续处理主外键关联时特别有用，省一次多余的查询。

批量插入是另一个高频需求，写法跟单条插入的区别就在 executemany。

data = [ ("user1", "user1@example.com", 20), ("user2", "user2@example.com", 21), ("user3", "user3@example.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 '1'='1" sql = f"SELECT * 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（可重复读），具体有四种级别，按隔离强度从低到高排列如下。

这几种异常的通俗解释是这样的。脏读，就是事务 A 读到事务 B 还没提交的数据，结果 B 回滚了，A 读到的就是无效数据。不可重复读，事务 A 里同一查询执行两次，结果不一样因为其他事务在这期间提交了修改。幻读，事务 A 按条件查出来一批行，期间事务 B 插入了新行，A 再查多出来了几条。

InnoDB 在 REPEATABLE READ 级别下通过 MVCC 和间隙锁基本解决了幻读问题，这也是它能成为默认级别的原因。Python 层面设置隔离级别的方法如下。

conn = pymysql.connect( host="127.0.0.1", user="root", password="your_password", database="demo", charset="utf8mb4" ) 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( creator=pymysql, maxconnections=20, mincached=5, maxcached=10, maxshared=10, blocking=True, maxusage=None, setsession=[], ping=1, host="127.0.0.1", port=3306, user="root", password="your_password", database="demo", charset="utf8mb4", cursorclass=pymysql.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，连接耗尽时新请求阻塞等待而不是直接报错。ping=1 表示取连接时如果连接空闲超过一定时间就发送 ping 包检查连接是否有效，能避免拿到数据库已断开但应用不知道的死连接。

有个坑我要重点提示，从连接池拿连接用完一定记得 close，但这个 close 不是真的关闭连接，而是归还给池子。如果忘了这一步，连接会被一直占用，最后池子耗尽，整个应用卡死。

4.3 生产环境连接池方案对比与选型

DBUtils 的 PooledDB 够用，但它的连接池是单进程内的。如果应用是多进程部署，每个进程都有自己的池子，总连接数是进程数乘池大小，规划时要把这个乘数算进去。

实际生产里我见过更主流的方案，是用 SQLAlchemy 作为 ORM 层，它内部自带连接池，而且对 PyMySQL 和 mysqlclient 做了统一封装，切驱动只改一个 URL 就行。

from sqlalchemy import create_engine engine = create_engine( "mysql+pymysql://root:your_password@127.0.0.1:3306/demo?charset=utf8mb4", pool_size=10, max_overflow=5, pool_pre_ping=True, pool_recycle=3600 )

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(host="127.0.0.1", user="root", password="your_password", database="demo", charset="utf8mb4") cursor = conn.cursor() rows = [(f"user_{i}", f"user_{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( host="127.0.0.1", user="root", password="your_password", database="demo", charset="utf8mb4", read_timeout=10, write_timeout=10, autocommit=False )

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 问题速查表

我整理了一下实际项目里出现频率最高的问题和解决方案。

6.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 'root'@'localhost' 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 就是专门解决这个问题的，一定不要省。