1. 为什么你的异步 MySQL 代码总是卡在连接池上aiomysql 是 Python 生态里做异步 MySQL 访问最常用的驱动它把 PyMySQL 的协议实现和 asyncio 的事件循环结合起来让你在协程里直接await数据库操作不用再靠线程池硬扛。适合谁适合已经在用 FastAPI、Sanic、aiohttp 这类异步框架或者准备把现有同步 DB 调用改造成异步链路的 Python 后端开发者。它能做什么一句话用连接池管理长连接、用协程并发跑 CRUD、用事务保证多表写入的一致性。但实际写起来很多人第一步就踩坑连接池在asyncio.run()里创建函数一退出池子就关了或者autocommit没配对插入数据查不到再或者把连接池对象当全局变量跨事件循环复用直接报Event loop is closed。这篇教程就围绕 aiomysql 连接池初始化、CRUD 封装、配置文件骨架三件事展开同时把 TaoToken 统一 Key 的接入方式串进去——你不需要在代码里散落一堆模型服务的密钥用一个 Key 走统一入口就行。下面所有代码都可以直接复制到本地跑通目标是在你自己的机器上完成一次异步读写冒烟测试。2. TaoToken 前置统一 Key 与配置文件骨架TaoToken 在这里的角色是统一凭证入口。你注册后拿到一个 API Key后续无论是调模型对话、跑 Coding Plan 还是访问兼容接口都用同一个 Key不用每个服务单独配一套密钥。官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 注意 API 地址不带 UTM 参数。先把配置骨架搭好。我习惯用config.toml存数据库连接信息用环境变量存 Key两者分开避免把密钥写进版本库。# config.toml [mysql] host 127.0.0.1 port 3306 user root password your_mysql_password db aiomysql_demo charset utf8mb4 autocommit false [mysql.pool] minsize 1 maxsize 10 pool_recycle 300 connect_timeout 10 [taotoken] base_url https://taotoken.net/api # api_key 从环境变量 TAOTOKEN_API_KEY 读取不写死在这里如果你更习惯 JSON等价骨架如下{ mysql: { host: 127.0.0.1, port: 3306, user: root, password: your_mysql_password, db: aiomysql_demo, charset: utf8mb4, autocommit: false, pool: { minsize: 1, maxsize: 10, pool_recycle: 300, connect_timeout: 10 } }, taotoken: { base_url: https://taotoken.net/api } }读取配置用 Python 3.11 之后内置的tomllib不用额外装包import tomllib import os with open(config.toml, rb) as f: cfg tomllib.load(f) MYSQL_CFG cfg[mysql] POOL_CFG cfg[mysql][pool] TAOTOKEN_KEY os.environ[TAOTOKEN_API_KEY] TAOTOKEN_BASE cfg[taotoken][base_url]注意autocommit建议保持false由代码显式commit这样事务边界清晰出问题也好回滚。如果你确实想每条语句自动提交再改成true。Key 的获取和查看在控制台的 API Keys 页面https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。拿到后写进环境变量别硬编码。3. 可复制配置aiomysql 连接池初始化与 CRUD 封装3.1 建库建表 SQL先准备测试表用户表和订单表覆盖主键、唯一约束、外键和索引CREATE DATABASE IF NOT EXISTS aiomysql_demo DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE aiomysql_demo; CREATE TABLE IF NOT EXISTS user ( id INT UNSIGNED NOT NULL AUTO_INCREMENT, username VARCHAR(50) NOT NULL, age TINYINT UNSIGNED DEFAULT 0, email VARCHAR(100) NOT NULL, create_time DATETIME DEFAULT CURRENT_TIMESTAMP, update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_email (email) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE IF NOT EXISTS order ( order_id INT UNSIGNED NOT NULL AUTO_INCREMENT, user_id INT UNSIGNED NOT NULL, order_no VARCHAR(32) NOT NULL, amount DECIMAL(10,2) NOT NULL DEFAULT 0.00, create_time DATETIME DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (order_id), UNIQUE KEY uk_order_no (order_no), KEY idx_user_id (user_id), CONSTRAINT fk_order_user FOREIGN KEY (user_id) REFERENCES user (id) ON DELETE CASCADE ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;3.2 连接池初始化连接池是整个异步链路的核心。用aiomysql.create_pool创建参数从配置读import asyncio import aiomysql from aiomysql import DictCursor _pool None async def init_pool(): global _pool if _pool is None: _pool await aiomysql.create_pool( hostMYSQL_CFG[host], portMYSQL_CFG[port], userMYSQL_CFG[user], passwordMYSQL_CFG[password], dbMYSQL_CFG[db], charsetMYSQL_CFG[charset], autocommitMYSQL_CFG[autocommit], minsizePOOL_CFG[minsize], maxsizePOOL_CFG[maxsize], pool_recyclePOOL_CFG[pool_recycle], connect_timeoutPOOL_CFG[connect_timeout], cursorclassDictCursor, ) return _pool async def close_pool(): global _pool if _pool: _pool.close() await _pool.wait_closed() _pool Nonepool_recycle300是关键参数MySQL 默认 8 小时断开空闲连接设成 300 秒让池子主动回收避免拿到失效连接报Lost connection。3.3 CRUD 封装把增删改查封成独立协程统一从池子取连接async def insert_user(username: str, age: int, email: str) - int: pool await init_pool() async with pool.acquire() as conn: async with conn.cursor() as cur: sql INSERT INTO user (username, age, email) VALUES (%s, %s, %s) await cur.execute(sql, (username, age, email)) await conn.commit() return cur.lastrowid async def query_user(user_id: int): pool await init_pool() async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute( SELECT id, username, age, email FROM user WHERE id %s, (user_id,), ) return await cur.fetchone() async def update_user_age(user_id: int, new_age: int) - int: pool await init_pool() async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute( UPDATE user SET age %s WHERE id %s, (new_age, user_id), ) await conn.commit() return cur.rowcount async def delete_user(user_id: int) - int: pool await init_pool() async with pool.acquire() as conn: async with conn.cursor() as cur: await cur.execute(DELETE FROM user WHERE id %s, (user_id,)) await conn.commit() return cur.rowcount参数化查询用%s占位别用字符串拼接这是防 SQL 注入的底线。3.4 事务封装多表写入必须走事务出错回滚async def create_user_with_order(username: str, email: str, order_no: str, amount: float): pool await init_pool() async with pool.acquire() as conn: async with conn.cursor() as cur: try: await cur.execute( INSERT INTO user (username, email) VALUES (%s, %s), (username, email), ) user_id cur.lastrowid await cur.execute( INSERT INTO order (user_id, order_no, amount) VALUES (%s, %s, %s), (user_id, order_no, amount), ) await conn.commit() return user_id except Exception: await conn.rollback() raise4. 验证请求跑通异步读写冒烟测试把上面的代码拼成一个可执行脚本验证整条链路async def smoke_test(): await init_pool() uid await insert_user(张三, 20, zhangsantest.com) print(insert ok, id , uid) row await query_user(uid) print(query ok, row , row) affected await update_user_age(uid, 22) print(update ok, affected , affected) row await query_user(uid) assert row[age] 22, age 未更新 deleted await delete_user(uid) print(delete ok, affected , deleted) await close_pool() print(smoke test passed) if __name__ __main__: asyncio.run(smoke_test())预期输出insert ok, id 1 query ok, row {id: 1, username: 张三, age: 20, email: zhangsantest.com} update ok, affected 1 delete ok, affected 1 smoke test passed如果这一步跑通说明连接池、CRUD、事务提交都没问题。接下来验证并发场景用asyncio.gather同时跑多个查询async def concurrent_test(): await init_pool() ids [await insert_user(fu{i}, 20 i, fu{i}test.com) for i in range(5)] results await asyncio.gather(*[query_user(i) for i in ids]) for r in results: print(r[username], r[age]) await close_pool() asyncio.run(concurrent_test())连接池会自动分配连接maxsize10时 5 个并发任务不会互相阻塞。5. 本篇常见错排查5.1 Event loop is closed报错长这样RuntimeError: Event loop is closed。原因通常是连接池在asyncio.run()里创建函数退出后事件循环关闭但池子对象还被全局引用下次再asyncio.run()时复用旧池子。解决办法每次asyncio.run()内部重新init_pool()退出前close_pool()别跨事件循环复用。5.2 Lost connection to MySQL server空闲连接被 MySQL 服务端断开。检查pool_recycle是否设置建议 300 秒以内。另外connect_timeout别设太小网络抖动时容易误判。5.3 插入成功但查不到数据autocommitfalse时忘了await conn.commit()。所有写操作后必须显式提交否则连接归还池子时事务回滚。5.4 IntegrityError: Duplicate entry唯一约束冲突比如邮箱重复。捕获aiomysql.IntegrityError单独处理别让它冒泡到顶层from aiomysql import IntegrityError try: await insert_user(李四, 25, zhangsantest.com) except IntegrityError as e: print(邮箱已存在:, e)5.5 连接池参数验证minsize别大于maxsize否则创建池子直接报错。maxsize根据 MySQL 的max_connections和你的并发量调本地开发 10 够用生产环境按 QPS 估算。验证参数是否生效pool await init_pool() print(pool size:, pool.size, free:, pool.freesize)pool.size是当前连接数pool.freesize是空闲连接数跑并发任务时观察这两个值变化确认池子在正常工作。6. 把统一 Key 接进你的异步链路数据库链路跑通后如果你还要在同一个项目里调模型做数据处理、生成 SQL 或者做结果摘要TaoToken 的统一 Key 可以直接复用。模型对话入口在 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 长期跑编码任务或 Agent 场景可以看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API Keys 管理在 https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。一个 Key 走完数据库和模型两条链路配置文件里只留一个环境变量代码里不用散落多套凭证。先把smoke_test()跑绿再把并发测试跑一遍你的异步 MySQL 链路就算真正落地了。