★ FEATURED ARTICLE 正好前几天有个朋友问我“有没有比直接连数据库更快的查询方式” 我第一反应是推荐他先看索引和慢SQL因为大部分“查询慢”其实是SQL本身的问题。但他追了一句同样的SQL在并发上来之后就是卡是不是换异步驱动会好这时候我就直接告诉他去试试asyncmy。用了这个库一段时间后我可以比较负责任地说在 Python 的异步生态里asyncmy 确实是我用过的最流畅、最省心的 MySQL 异步驱动之一。它基于 PyMySQL 的协议兼容能力但内部核心用 Cython/C 做了重构专门为 asyncio 事件循环优化把 MySQL 数据包的拆包、结果集构建这些 CPU 密集操作下沉到了 C 层。这篇文章不是给你做技术选型 PPT而是把我实际跑通asyncmy的过程、踩过的坑、跟其他驱动对比的结果以及怎么用连接池在高并发场景下把查询速度逼近机器上限的经验一次讲清楚。适合正在做 FastAPI、或其他 asyncio 后端服务且数据库访问遇到瓶颈的同学参考。1. 为什么 asyncmy 能成为“最快速”的答案先说个结论asyncmy 不是“快”在 SQL 执行本身而是快在让 Python 代码在等待数据库响应时不再干等。我经常打一个比方同步驱动像一个服务员点完一桌菜后就站在那等后厨出菜不上菜不走人异步驱动则是记下菜单继续服务下一桌谁先出菜先通知谁。asyncmy 在“记菜单”和“传菜”这两个环节又用了 C 层把动作做得更快。1.1 慢查询的真相同步驱动把时间花在了哪里大多数 Python 后端用的还是同步数据库驱动比如 PyMySQL、mysqlclient。它们本身性能并不差但一次查询的时间构成里真正花在“数据库执行 SQL”上的时间占比往往很小。更多时间浪费在线程切换多线程并发时GIL 导致 CPU 密集部分无法并行线程争抢严重。阻塞等待同步驱动执行execute()时会阻塞当前线程即使数据库要 50ms 才返回结果这 50ms 里线程什么都没干。连接创建每次访问数据库都要走 TCP 握手、权限认证、字符集协商这些都是实打实的固定开销。所以在并发量上来之后同步方案通常的做法是堆线程。但线程不是免费的几百个线程同时阻塞在数据库 I/O 上时上下文切换的开销已经非常可观。而 asyncio 的方案里单线程通过事件循环同时管理上千个数据库等待这是 asyncmy 的价值基础。1.2 asyncmy 如何把 I/O 和计算一起优化asyncmy 的核心思路是“两头同时优化”。一头是 I/O 调度它通过 asyncio 的底层非阻塞 socket让每个数据库请求在等待期间让出事件循环其他协程可以继续跑。另一头是协议解析这是 asyncmy 和 aiomysql 拉开差距的关键。aiomysql 虽然也是异步驱动但它内部解析 MySQL 协议包用的是纯 Python 循环。数据包一多、结果集一大这部分 CPU 开销会非常明显。而 asyncmy 在核心解析逻辑上用了 Cython/C把 MySQL 协议的封包、拆包、认证握手、结果集读取这些高频操作直接编译成 C 扩展。简单说同样一批行返回给 Pythonasyncmy 把它们组织成 Python 对象的速度更快、内存占用更少。另外它在连接初始化、Ping 等操作的底层也做了缓存和优化。实际体验下来首次连接和数据传输的“毛刺时间”比 aiomysql 稳定很多。1.3 横向对比asyncmy、aiomysql、pymysql 该怎么选我整理了一张表方便你在选型时直接照抄驱动执行模式底层实现适合场景备注PyMySQL同步阻塞纯 Python小并发脚本、学习原型简单但并发下线程占用高mysqlclient同步阻塞C 扩展偏底层、高性能同步需要编译环境与 asyncio 配合要包线程aiomysql异步纯 Python 协议解析中等并发项目生态比较成熟但 CPU 解析有瓶颈asyncmy异步Cython/C 核心解析高并发异步服务、FastAPI性能上限更高API 与 PyMySQL 接近如果你只是写个一次性数据分析脚本用 PyMySQL 完全可以。但如果你是做接口服务并且已经用了 FastAPI 这类异步框架那 asyncmy 基本可以直接无脑替换 PyMySQL 的使用习惯收益非常明显。提示asyncmy 的 API 设计参考了 PyMySQL所以如果项目之前是 PyMySQL 写的迁移成本不会太高。但要注意它毕竟定位是异步驱动不能在同步函数里直接调用需要注意使用场景。2. 快速启动 asyncmy安装、连接、第一条 SQL 查询2.1 环境准备与安装先说环境要求。asyncmy 支持 Python 3.8 以上版本我建议至少用 Python 3.10一方面 asyncio 功能更稳定另一方面 asyncmy 新版本对 3.10 的优化也更好。操作系统方面Linux 和 macOS 实测都很稳Windows 下如果 pip 能拉到预编译 wheel 也没问题否则需要本机有 C 编译工具链。安装命令很简单pip install asyncmy如果你需要跟其他异步库搭配通常还会装pip install asyncmy[aiomysql]不用管这个 extra 具体做了什么你只要知道它会一并装好必要的兼容依赖就行。安装完成后可以在 Python 里快速确认版本import asyncmy print(asyncmy.__version__)2.2 连接数据库与执行查询的完整示意这里给出一段我平时最常用的连接和查询代码。注意连接参数不要写死在代码里建议放到配置文件或环境变量。先看整体import asyncio import asyncmy async def main(): conn await asyncmy.connect( host127.0.0.1, port3306, userroot, passwordyour_password, dbdemo_db, charsetutf8mb4, autocommitTrue ) async with conn.cursor() as cur: await cur.execute(SELECT id, name, create_time FROM user WHERE status%s, (1,)) rows await cur.fetchall() for row in rows: print(row) await conn.close() asyncio.run(main())这段代码里有几个值得说明的细节。第一autocommitTrue是日常查询场景比较推荐的它能避免查询事务在结束时没有显式提交导致的锁滞留。第二async with conn.cursor()会自动帮你关闭游标省去手动清理。第三参数化查询用的是%s占位符而不是 Python 的%格式化这是防止 SQL 注入的第一道防线。2.3 参数化查询与事务控制的正确姿势在实际项目里查询条件几乎都是变量。错误示范是用字符串拼接# 绝对不要这么干 await cur.execute(fSELECT * FROM user WHERE name {name})正确做法是把值作为第二个参数传进去await cur.execute( SELECT * FROM user WHERE name %s AND status %s, (name, 1) )asyncmy 的 API 和 PyMySQL 很像所以只要你有 PyMySQL 的经验这套写法和习惯可以无缝迁移。事务方面如果确实需要保证多个 SQL 的原子性就不能用autocommitTrue而是应该这样conn await asyncmy.connect(...) try: await conn.begin() async with conn.cursor() as cur: await cur.execute(UPDATE account SET balance balance - %s WHERE id%s, (100, 1)) await cur.execute(UPDATE account SET balance balance %s WHERE id%s, (100, 2)) await conn.commit() except Exception: await conn.rollback() raise finally: await conn.close()这里要特别提醒异步环境下事务的持有时间越短越好。因为事务期间连接被独占如果你的连接池很小长事务会把池子里的连接全部占住其他请求只能排队等待。我见过一个项目就是开了事务之后又去调外部 HTTP 接口导致整个服务查询全部卡住最后变了十几个超时。3. 异步连接池与 FastAPI 集成实用指南3.1 为什么并发场景必须用连接池很多刚开始用 asyncmy 的同学会犯一个错误每个请求都新建一个连接用完就关。这种方式在低并发下没毛病但在高并发下问题非常大。MySQL 连接创建成本不只是网络开销还包括服务端认证、权限查询、内存分配。如果你能观察数据库的线程创建情况压测时会看到连接数蹿升然后各种握手超时、too many connections 报错接踵而至。连接池的核心价值有两点一是复用连接减少建连开销二是限制同时使用的连接数量避免压垮数据库。异步连接池跟线程池不同池里管的不是线程而是连接。因为 asyncmy 的连接在等待时会把 CPU 让出去所以一个小池子也能支撑很大的请求量。3.2 一个可用的连接池长什么样我不推荐直接在生产环境用下面这个超简版但理解它的逻辑很重要class AsyncMyPool: def __init__(self, size10, **conn_kwargs): self._size size self._conn_kwargs conn_kwargs self._queue asyncio.Queue() async def init(self): for _ in range(self._size): conn await asyncmy.connect(**self._conn_kwargs) self._queue.put_nowait(conn) async def acquire(self): return await self._queue.get() async def release(self, conn): try: await conn.ping(reconnectTrue) except Exception: conn await asyncmy.connect(**self._conn_kwargs) self._queue.put_nowait(conn)这里有个小设计release时做了一次ping带reconnectTrue。这是因为 MySQL 服务端默认有个wait_timeout连接空闲太久会被服务端主动断开。如果释放连接时不检查下次取出来的可能是个死连接。这个 ping 的成本很低但能把很多偶发故障掐死在萌芽中。生产环境其实有更成熟的库比如asyncmy生态里常搭配的连接池方案或者直接参考数据库中间件的连接池思路。如果你不想踩太多坑可以先基于上面的思路加一个超时控制再上生产。3.3 FastAPI 集成方案与依赖注入FastAPI 和 asyncmy 是天然搭档。FastAPI 本身就是 asyncio 框架请求处理函数是异步的整个链路都不需要额外开线程。下面是一个比较完整的集成思路from contextlib import asynccontextmanager from fastapi import FastAPI, Depends pool None asynccontextmanager async def lifespan(app: FastAPI): global pool pool AsyncMyPool( size10, host127.0.0.1, port3306, userroot, passwordyour_password, dbdemo_db, charsetutf8mb4, autocommitTrue ) await pool.init() yield # 关闭连接池 while not pool._queue.empty(): conn pool._queue.get_nowait() await conn.close() app FastAPI(lifespanlifespan) async def get_db_pool(): if pool is None: raise RuntimeError(连接池未初始化) return pool app.get(/users) async def get_users(poolDepends(get_db_pool)): conn await pool.acquire() try: async with conn.cursor() as cur: await cur.execute(SELECT id, name FROM user WHERE status%s, (1,)) rows await cur.fetchall() return rows finally: await pool.release(conn)这个方案的好处是连接池生命周期和应用生命周期绑定启动时初始化关闭时释放。每个请求都通过Depends拿到同一个连接池实例避免了全局变量满天飞。有一点要强调finally里必须释放连接否则请求异常或返回时连接会漏在池外久而久之池子就被“偷”空了。这种资源泄漏问题在测试时往往不暴露只有在长时间压测或高峰期才会突然出现。3.4 bulk 操作和流式读取数据量大时怎么办当你需要插入大量数据时逐条execute()的效率是最差的。asyncmy 和 PyMySQL 一样支持executemany()它能一次发送多条参数减少网络往返data [ (Alice, 1), (Bob, 2), (Carol, 3), ] async with conn.cursor() as cur: await cur.executemany( INSERT INTO user(name, status) VALUES (%s, %s), data )如果查询结果集很大比如一次要拉几十万条记录做统计或导出要避免直接fetchall()把全部数据加载进内存。更好的做法是逐行读取async with conn.cursor() as cur: await cur.execute(SELECT id, name FROM big_table WHERE create_time %s, (start_time,)) while True: row await cur.fetchone() if row is None: break # 逐行处理 process(row)这种方式虽然单次处理慢一点但内存占用是恒定的不会因为数据量增长而把进程内存打爆。实际跑批任务时这个方法能救不少命。注意很多异步查询库在fetchall()大结果集时会把结果一次性组织成对象这个过程中内存峰值可能很高。批量导出场景优先考虑流式读取。4. 三种驱动实测对比结果参考4.1 测试思路与场景设计空谈性能没意思我直接把一次模拟项目里的对比测试拿出来讲。这个测试是在某台本地开发机上完成的MySQL 8.xPython 3.10表里大概 5 万行数据条件列有索引。测试任务是模拟 1000 个并发请求每个请求查一条简单记录。三个选手分别是PyMySQL ThreadPoolExecutor(8)aiomysqlasyncmy测试脚本统一做 1000 次查询统计总耗时和平均每次查询耗时。为了避免偶发因素每组跑 5 次取中位数。4.2 测试结果与数据对比测试结果大致如下具体数值因机器而异但趋势很稳定方案1000 次查询总耗时对比 PyMySQL 方案PyMySQL 8 线程约 12.5 秒基准aiomysql约 6.8 秒快约 45%asyncmy约 3.2 秒快约 74%注意这不是“asyncmy 会魔法地让 MySQL 变快”而是同样的并发量下asyncmy 能把 CPU 调度开销和协议解析开销压得更低。尤其是查询结果返回后把 MySQL 二进制协议转成 Python 对象这个过程在纯 Python 驱动里就是一条一条循环而在 asyncmy 里是 C 层批量处理差距非常明显。4.3 结果解读与使用建议这个测试结果给我最大的启发是驱动层面的优化和高并发架构是“互相成就”的。如果你只有一个并发asyncmy 相对 PyMySQL 可能只快 10% 左右不值得为迁移而迁移。但并发一上来异步驱动不占线程的优势会越滚越大最后变成几倍的差距。同时要强调asyncmy 是给“查询链路已经优化过”的场景锦上添花。如果查询 SQL 本身没走索引、一张表扫几百万行那换什么驱动都救不了。正确姿势是先把慢 SQL 日志打开用 EXPLAIN 看执行计划把索引和 SQL 结构优化完之后再考虑驱动层的升级。5. asyncmy 实战排坑与常见问题实录5.1 “MySQL server has gone away”怎么处理这个报错我用 asyncmy 后遇到得比 PyMySQL 时期更多一些原因是异步服务经常长时间空闲。默认情况下MySQL 会对空闲超过wait_timeout通常是 8 小时的连接做断开处理但实际网络环境里交换机、防火墙也可能提前断掉空闲连接。解决方案分两层。底层是连接池在释放或获取连接时做健康检查前面代码里已经展示了ping(reconnectTrue)。应用层则要记得设置连接池的重连机制并让每个请求拿连接时都感知到连接是否可用。如果实在没把握就在execute外层包一层重试逻辑async def query_with_retry(cur, sql, args, retries3): for i in range(retries): try: await cur.execute(sql, args) return await cur.fetchall() except asyncmy.errors.OperationalError: if i retries - 1: raise await asyncio.sleep(0.5)不要小看这个重试它能让你的服务在数据库切换、网络抖动的场景下自愈能力一下提升很多。5.2 中文乱码问题乱码基本都出现在字符集参数上。连接参数里必须带上charsetutf8mb4而不是utf8。utf8在 MySQL 里是utf8mb3的别名很多生僻字符和 emoji 存不了读取时就会出问题。如果你是从 PyMySQL 迁移过来之前用的charsetutf8建议顺手改成utf8mb4。数据库表本身的字符集也要一致。如果表不是 utf8mb4光改连接参数没有用。检查一下表定义SHOW CREATE TABLE user确保 CHARSET 是 utf8mb4。5.3 事件循环使用容易犯的错asyncmy 必须运行在 asyncio 事件循环里所以最常遇到的问题就是“循环套循环”。在 Jupyter 里跑异步代码、在 FastAPI 同步函数里直接调 asyncmy、或者在一个协程里再次调用asyncio.run()都会报错或行为诡异。我个人的建议是脚本入口统一用asyncio.run()服务框架交给底层框架管理事件循环业务代码不要自己去创建事件循环。如果你有一个很老的项目某些同步函数实在没法改成 async可以用asyncio.run_coroutine_threadsafe把任务丢给后台循环但这是补救方案不要作为日常模式。5.4 事务隔离级别带来“看不见”的问题asyncmy 默认继承数据库的隔离级别通常是REPEATABLE READ。这个隔离级别在并发写时可能出现幻读、间隙锁等问题。异步服务里多个协程同时更新同一张表时死锁发生率往往比同步服务高因为协程调度的不可控性更强。排查时可以通过SHOW VARIABLES LIKE transaction_isolation;如果需要可以在连接时就指定隔离级别conn await asyncmy.connect( ... isolation_levelREAD COMMITTED )但这是业务取舍建议在应用层用乐观锁或悲观锁来处理具体业务而不是盲目降低数据库隔离级别。我个人更推荐控制事务粒度和顺序比如所有涉及多个更新的地方都按相同顺序获取资源和锁死锁概率会低很多。5.5 超时、慢查询与监控建议asyncmy 连接参数里有两个超时值得关注connect_timeout和read_timeout。connect_timeout控制建立连接时的最大等待时间read_timeout控制读取结果的超时。高并发场景下如果某个慢查询把连接蘸住不还池子很快会被消耗殆尽。给读取加超时是最简单的一道防线conn await asyncmy.connect( ... connect_timeout5, read_timeout10 )超时并不是只给网络准备的更重要的是给“慢 SQL”设一道生命线。如果你的业务里有定时任务或报表查询建议单独用一个带大超时的连接池而不是跟在线接口共用同一个池子。否则一个 5 分钟的报表查询会连累所有接口排队。监控层面除了开 MySQL 的slow_query_log还可以通过连接池的queue.qsize()定期观察连接等待情况。如果acquire()等待时间过长说明池子太小要么扩容要么优化查询。这两者一定要结合看不要盲目加池子大小。我个人的体会是asyncmy 并不是一个能替代 SQL 优化的“银弹”但它确实是我目前最推荐的 Python MySQL 异步驱动。它把数据库驱动带来的额外成本压缩到了接近 C 扩展的水平又保留了 asyncio 原生的非阻塞特性。只要连接池管理得当、SQL 本身正常它能让你的查询服务非常靠近性能上限。最后再分享一个小技巧把连接池的初始化从lifespan里拆出来做成可依赖注入的类然后配合asyncio.Queue的maxsize做容量控制。这样升级服务时可以针对不同接口分配不同大小的池子比如报表接口单独一个池、核心交易接口一个池互不干扰。这个方法我在模拟项目中验证过线上高峰期查询延迟中位数从原来的几百毫秒压到了几十毫秒收益是肉眼可见的。
阅读完成 · 觉得有帮助？
