企业官网建设流程全解析

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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_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`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; 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 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

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( host=MYSQL_CFG["host"], port=MYSQL_CFG["port"], user=MYSQL_CFG["user"], password=MYSQL_CFG["password"], db=MYSQL_CFG["db"], charset=MYSQL_CFG["charset"], autocommit=MYSQL_CFG["autocommit"], minsize=POOL_CFG["minsize"], maxsize=POOL_CFG["maxsize"], pool_recycle=POOL_CFG["pool_recycle"], connect_timeout=POOL_CFG["connect_timeout"], cursorclass=DictCursor, ) return _pool async def close_pool(): global _pool if _pool: _pool.close() await _pool.wait_closed() _pool = None

pool_recycle=300是关键参数，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() raise

4. 验证请求：跑通异步读写冒烟测试

把上面的代码拼成一个可执行脚本，验证整条链路：

async def smoke_test(): await init_pool() uid = await insert_user("张三", 20, "zhangsan@test.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': 'zhangsan@test.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(f"u{i}", 20 + i, f"u{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())

连接池会自动分配连接，maxsize=10时 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 插入成功但查不到数据

autocommit=false时忘了await conn.commit()。所有写操作后必须显式提交，否则连接归还池子时事务回滚。

5.4 IntegrityError: Duplicate entry

唯一约束冲突，比如邮箱重复。捕获aiomysql.IntegrityError单独处理，别让它冒泡到顶层：

from aiomysql import IntegrityError try: await insert_user("李四", 25, "zhangsan@test.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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，长期跑编码任务或 Agent 场景可以看 Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API Keys 管理在 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。

一个 Key 走完数据库和模型两条链路，配置文件里只留一个环境变量，代码里不用散落多套凭证。先把smoke_test()跑绿，再把并发测试跑一遍，你的异步 MySQL 链路就算真正落地了。