用了 asyncpg 小半年从最开始只是图它原生支持 asyncio到后来因为性能瓶颈不得不把它从驱动程序列表里单独拎出来研究我对这个库的理解经历了“会用”到“用对”两个阶段。今天这篇就把这两层东西都写透先聊 asyncpg 到底解决什么问题为什么它在 asyncio 生态里几乎是绕不开的选择再拆它内部几个关键机制最后给一套可以直接抄的实操代码和排坑手册。写给你那些马上要在 FastAPI、爬虫、任务管道里用 Postgres又不想被同步驱动拖垮的 Python 后端开发者。1. 为什么是 asyncpg而不是 psycopg2 加线程池1.1 asyncio 生态里的天然缺口这几年 Python 异步技术栈基本稳定在 asyncio 这一套事件循环上Web 框架这边有 FastAPI、Sanic、Quart消息消费那边有 aiokafka、aio-pika但你仔细看就能发现数据库驱动这块历史包袱最重。传统系统里的主力驱动 psycopg2 是纯同步阻塞的一个cursor.execute()会把当前线程卡死直到数据库返回结果。在 asyncio 项目里直接用它等于把整个事件循环堵住并发全废。有人会说那不还有 psycopg3 吗psycopg3 确实支持异步但它的异步是建立在独立的线程或进程池上本质上还是把同步调用包装成协程属于“模拟异步”和 asyncpg 这种直接从协议层就为异步设计的驱动完全是两码事。asyncpg 自己用 Cython 实现了 PostgreSQL 的 wire protocol收发数据走的是 asyncio 的事件循环没有线程切换也没有上下文切换的开销。用一句话概括asyncpg 是真正意义上“长在 asyncio 树上的驱动”而不是“给同步驱动穿了一件协程外套”。1.2 和同类驱动放一起看差距到底在哪我把 psycopg2 线程池、psycopg3 异步模式、asyncpg 三种方案在本地同一个数据库上跑过一轮压测场景是 5000 条简单查询每条只查主键。结果非常直观asyncpg 的吞吐量大约是 psycopg2 线程池方案的 2 到 3 倍延迟的 P99 也明显更低。这还只是简单查询如果涉及到批量写入、COPY 导入差距还会继续拉大。不过这不是说让你无脑全面替换。如果你的项目重度依赖 SQLAlchemy 的 ORM 自动映射和方言兼容先别急SQLAlchemy 1.4 之后提供了asyncpg作为 async 方言你照样可以create_async_engine(postgresqlasyncpg://...)。所以现在这个阶段用 asyncpg 不意味着放弃 SQLAlchemy而是给 ORM 换了一颗更强劲的心脏。对比表可以这样看方案异步方式协议实现典型场景踩坑指数psycopg2 ThreadPoolExecutor线程池模拟libpq 封装存量项目、重度 ORM中等psycopg3 async等待回调libpq 封装需要兼容 psycopg2 API 的项目中等asyncpg原生协程Cython 自研协议高并发、追求极致性能略高1.3 它到底解决了什么问题说穿了asyncpg 解决的三个核心痛点第一是并发效率一个进程里起几百个协程同时查库不阻塞第二是连接资源管理通过内置连接池优雅地复用数据库连接避免每次请求都握手鉴权第三是数据转换速度PG 的二进制协议解析在 asyncpg 里被压榨到了极致减少不必要的一次次字符串编解码。如果你只记得一句话那就是在 Python 异步世界里要跟 PostgreSQL 打交道asyncpg 是你绕不开的名字它既是性能上限也是很多诡异 bug 的根源。2. 深度拆解连接、预处理语句和类型转换机制2.1 连接生命周期远比你想象的复杂asyncpg 的Connection对象底层对应一条真实的 PostgreSQL 连接创建代价很高包括 TCP 握手、鉴权、参数协商。所以我反复强调生产环境不要用asyncpg.connect()一下连一个用完就断正确姿势一定是走连接池。连接池里的连接是有生命周期的不是说你把它放回去就万事大吉。数据库端可能有statement_timeout、空闲会话回收策略网络层可能有防火墙断空闲连接这些都会让你的池子里的“僵尸连接”变多。asyncpg 的连接池参数里有个max_inactive_connection_lifetime默认 300 秒超过这个时间的空闲连接会被主动关掉并替换成新连接。如果你发现自己连接老是断先检查这个参数和数据库侧的 wait_timeout 是不是不匹配。实操上有几个常用参数值得盯紧min_size池子保底连接数建议大于等于你的业务并发波谷。max_size池子上限建议不要超过数据库max_connections的一半留出管理连接和运维操作的余量。command_timeout单条命令超时不是连接超时很多人搞混。timeout获取连接的等待超时。高并发下池子耗尽这里的报错是PoolAcquireTimeoutError。另外强烈建议在每次acquire()后使用async with而不是手动 release。手动 release 最容易犯的错是查询中途抛异常导致连接永远回不了池子最后池子被占满程序假死。你如果写的是async with pool.acquire() as conn: result await conn.fetch(SELECT ...)完全不用关心释放问题异常也会把连接还回去这是最稳妥的写法。2.2 预处理语句缓存性能的隐形翅膀每次执行 SQL数据库都需要经历解析、计划、执行三步。对于高频重复的查询如果每次都让数据库重新解析白白浪费 CPU。asyncpg 对这个问题的处理非常激进它会为每条 SQL 自动创建预处理语句并且默认缓存。这一点和 psycopg2 需要手动prepare完全不同你用conn.fetch()、conn.execute()时asyncpg 内置语句缓存已经生效。缓存 key 是完整 SQL 字符串所以如果你写 SQL 时喜欢拼接空格或者大小写习惯不一致会导致同一条语义完全相同的 SQL 在缓存里产生多个条目缓存命中率下降数据库端还攒了一堆预处理语句。踩过一次坑后我的规范是项目里所有 SQL 字符串统一走常量定义不允许飘字面量参数一律通过$1、$2占位符传入。这样既提升缓存命中率又顺手防了注入一箭双雕。还有一个细节asyncpg的预处理语句缓存是在连接级别的不是连接池级别。也就是说池子里 10 条连接同一 SQL 会被预处理 10 次这是设计使然不用纠结。你真正要避免的是在一条连接上交替执行大量“一次性查询”把缓存挤爆。statement_cache_size默认 100如果业务就是有上百条不同查询可以把这个值调大但不是越大越好每条预处理语句在数据库端都是有内存开销的。2.3 类型转换陷阱字符串、数组和自定义类型asyncpg 对 PostgreSQL 类型和 Python 类型的映射做过精心设计但最坑的恰恰是它的“太精心”。举个例子jsonb字段默认返回的是str不是dict。第一次用的时候我直接傻眼明明存进去的是对象读出来成了一个字符串。后来才知道asyncpg 对 json 的处理刻意保留原始文本避免隐式反序列化带来的性能损耗和潜在歧义。你需要显式指定 codec 或者自己json.loads()。看一下最常用的默认映射规则PostgreSQL 类型asyncpg 默认 Python 类型int2 / int4 / int8intfloat4 / float8floatnumericdecimal.Decimalboolbooltext / varcharstrjson / jsonbstr不做反序列化byteabytestimestamp / datedatetime 对象intervaldatetime.timedeltaarraylist元素按对应规则映射uuiduuid.UUIDrecord / 自定义类型asyncpg.Record 或需手动注册如果你的表里有自定义枚举类型或复合类型直接用默认驱动查询会直接抛异常告诉你找不到对应的编解码器。解决方法是给连接或连接池注册 codec或者干脆在 SQL 层面把自定义类型 cast 成基础类型比如enum_col::text。我喜欢后者改动小也方便迁移。2.4 连接池的正确打开方式连接池虽然不用你自己写但它的一堆参数和生命周期策略决定了你的服务在流量峰值是优雅降级还是直接雪崩。一个容易被忽略的点是asyncpg.create_pool默认是不做任何预热连接的。池子启动时不会立刻建连接通常要等第一个请求进来才建立。你可以显式设置min_size并调用一次pool.initialize()如果有的话或者干脆在应用启动阶段发一条SELECT 1把连接热起来。真实场景里冷启动碰上流量高峰数据库会一瞬间收到几十个连接创建请求后端的 auth 压力会让延迟毛刺非常明显提前热身是最便宜的大促保障。另一个是要知道连接池不是无限并发加速器。max_size设得越高数据库端的连接数就越多。很多线上故障都是因为连接数打到数据库max_connections上限连运维都登不进去。我没少看到同学把max_size设成 200然后数据库侧只有 100 连接上限服务一压测立刻雪崩。我的经验池子上限 数据库最大连接数 × 0.3 再除以实例数先保守后放开。3. 实操过程从建池到事务与批量导入的完整落地3.1 搭建一个可复用的异步数据访问模块不废话直接上代码。先写一个全局唯一的连接池管理模块import asyncpg from functools import lru_cache DATABASE_URL postgresql://user:pass127.0.0.1:5432/myapp lru_cache(maxsize1) def get_pool_factory(): return lambda: None # 占位后面用 async 单例 # 更实际的做法是封装成一个可等待的全局对象 class Database: _pool None classmethod async def init(cls, dsn: str DATABASE_URL): if cls._pool is None: cls._pool await asyncpg.create_pool( dsndsn, min_size5, max_size20, max_inactive_connection_lifetime60, command_timeout10, timeout5, server_settings{application_name: my_app}, ) return cls._pool classmethod async def close(cls): if cls._pool: await cls._pool.close() cls._pool None classmethod def pool(cls) - asyncpg.Pool: assert cls._pool is not None, 请先调用 await Database.init() return cls._pool async def fetch_user(user_id: int): sql SELECT id, name, email FROM users WHERE id $1 async with Database.pool().acquire() as conn: return await conn.fetchrow(sql, user_id)几个细节说一下。server_settings里设置application_name这个值会出现在 pg_stat_activity 里排障时一眼看出连接属于哪个服务。lru_cache(maxsize1)这个用法适合放一些不需要 await 的配置对象但连接池这种需要异步初始化的资源还是用类级别的懒加载单例更顺手。如果你在 FastAPI 里接入启动事件里调用init()关闭事件里调用close()中间处理函数直接调用fetch_user这种封装函数就好。不要在业务代码里到处create_pool那是灾难。3.2 事务、隔离级别和保存点asyncpg 的事务接口是 Python 异步库里做得最舒服的那一档直接用async with包一层async def transfer(from_id: int, to_id: int, amount: int): sql UPDATE accounts SET balance balance - $2 WHERE id $1 async with Database.pool().acquire() as conn: async with conn.transaction(): await conn.execute(sql, from_id, amount) await conn.execute(sql, to_id, -amount) # 如果这里抛出异常整个事务自动回滚conn.transaction()默认隔离级别是 PostgreSQL 的 READ COMMITTED如果你需要更严格的语义可以显式指定async with conn.transaction(isolationserializable): ...事务里还想打标记回滚到某个点用transaction()返回对象的savepoint()上下文async with conn.transaction() as tx: await conn.execute(INSERT INTO audit_log ...) async with tx.savepoint(): await conn.execute(INSERT INTO risky_table ...) # 只回滚到 savepoint不影响外部事务这个在实际业务里非常有用比如批量导入时一行脏数据不应该毁掉整个批次你可以在循环里给每一行一个 savepoint捕获异常后继续落下一行而不至于像普通事务那样全部回滚。事务里还有几个坑。第一事务内不能并发地使用同一个连接执行多条语句一条执行结束之前你不能在同一连接上再跑别的查询所以事务代码里如果出现asyncio.gather(conn.execute(...), conn.execute(...))那基本就是各种cannot perform operation: another operation is in progress报错的前兆。第二事务和连接池之间的交互特别注意连接在事务中被归还给池子手动 release是极端危险的操作务必让async with conn.transaction()和async with pool.acquire()在同一层嵌套且事务块先退出、连接再归还。async with Database.pool().acquire() as conn: async with conn.transaction(): ... # 事务已经提交/回滚连接此刻才安全归还嵌套顺序错了你的连接可能带着一个未提交事务回到池子下一个拿到这条连接的用户看到的是脏数据。3.3 批量写入execute_many、executemany 和 COPY 提速批量写入是 asyncpg 最能秀肌肉的地方。普通的循环单条execute是最差的方案每一次都要经过完整的预处理、发送、等待往返。改进版是executemanyasyncpg 接口里对应的是conn.executemany(sql, args_list)它会复用同一条 SQL批量发送参数。但注意executemany在官方文档里也承认目前还不是逐条预处理的最高效方式它会把参数拼接后按 SQL 协议发送比循环快但不一定比得上 COPY。真正的杀手锏是copy_records_to_table直接把内存里的记录流式写入表records [ (1, alice, 120), (2, bob, 80), (3, carol, 45), ] async with Database.pool().acquire() as conn: await conn.copy_records_to_table( user_balance, recordsrecords, columns(user_id, user_name, balance), timeout30, )实测数据导入 10 万条记录循环单条execute大约需要 30 到 40 秒executemany大概 3 到 5 秒copy_records_to_table只需要 1 秒左右。量越大COPY 优势越夸张。原因在于 COPY 是 PostgreSQL 的批量加载协议数据流式进入数据库少了每次执行的解析和计划开销也少了大量网络往返。COPY 模式下如果你需要数据转换、去重这类操作建议先落临时表再用一条 SQL 转到正式表。虽然多了一步但转型逻辑用 SQL 写清晰可控也不容易触到类型转换的边界问题。3.4 监听通知LISTEN/NOTIFY 的异步姿势很多人不知道asyncpg 还支持 PostgreSQL 的 LISTEN/NOTIFY 机制这在构建轻量级消息推送、缓存失效通知时特别好用。监听时连接不能做别的所以通常单独开一条连接async def listen_channel(channel: str, handler): conn await asyncpg.connect(DATABASE_URL) await conn.add_listener(channel, handler) try: while True: await asyncio.sleep(3600) finally: await conn.remove_listener(channel) await conn.close() def on_event(connection, pid, channel, payload): print(fchannel{channel}, payload{payload}) asyncio.create_task(listen_channel(user_updated, on_event))这个机制适合做小范围实时通知比如订单状态变化后让各业务节点感知。但毕竟是数据库层面的消息机制吞吐量远不能和 Redis Stream、RabbitMQ 相比别拿它当主力消息队列用切记。4. 常见问题与排查技巧实录4.1 连接池耗尽与连接泄漏这个是我见过最高频的生产事故。症状是服务开始大量抛asyncpg.pool.PoolAcquireTimeoutError接着整个服务 RT 飙升。排查路径分三步先看是不是泄漏。查SELECT * FROM pg_stat_activity WHERE application_name 你的应用如果连接数长期在max_size附近居高不下且大量连接状态是idle in transaction基本可以判定有事务没提交就还给了池子或者 acquire 后没 release。再看是不是短连接风暴。峰值流量打过来池子里的连接不够用新请求都在等待超时后抛错。这时候不是代码 bug而是容量规划问题。处理办法有两种一是提高max_size二是减少单条连接占用时间把慢查询降下来让连接快速周转。最后看池子参数是不是配置失误。我见过command_timeout0被当作“无限超时”用结果某条 SQL 卡死数据库连接资源被锁死整个池子被同一条慢查询拖垮。现在我的建议是任何线上服务都显式配置command_timeout至少 5 秒或 10 秒宁可超时报错也不要让连接无限期挂着。4.2 类型序列化和 codec 注册的门道常见报错是asyncpg.exceptions.DataError: invalid input for query argument ...或者cannot convert ... to type。原因基本是两个方向一是 Python 侧对象没有对应的 PG 类型二是自定义 PG 类型没有注册 codec。举个例子如果你用 numpy 的int64直接传参asyncpg 会拒绝解决方法是显式int(value)。如果查出来的jsonb默认是字符串你要的是 dict可以给连接注册 codecasync with Database.pool().acquire() as conn: await conn.set_type_codec( jsonb, encoderjson.dumps, decoderjson.loads, schemapg_catalog, ) row await conn.fetchrow(SELECT data::jsonb FROM t WHERE id$1, 1) # 此时 row[data] 是 dict不过连接池里的 codec 也是连接级别的只对当前获取到的这条连接生效。如果你想让池子里所有连接默认启用同一套 codec就得在每次 acquire 后设置或者包装一个初始化连接函数传给asyncpg.create_pool(init...)。init参数是个不错的方案它会在每条新连接建立时被调用统一注册 codec 和运行时参数。还有一类坑是时间格式。asyncpg 返回的timestamp是带时区本地化的datetime对象如果你和 Django TimeZone 或前端 JSON 相互转换务必统一成 UTC。我习惯在写入前用datetime.now(timezone.utc)生成读取后统一astimezone(timezone.utc)转成 ISO 字符串给前端省去一堆时区脑残 bug。4.3 慢查询排查与性能压测建议asyncpg 的日志默认不打印 SQL。排查慢查询时我一般是两条路并行一是开启 PostgreSQL 的auto_explain或pg_stat_statements直接看数据库侧的真实执行计划二是在应用侧给查询包一层计时把超过阈值的 SQL 记录到日志。第二条更直接能精确对应到代码位置。性能压测上我的经验是不要只测 QPS要看 P99 尾巴。asyncpg 的异步特性决定了它在平均延迟上很好看但如果你的某个查询手段不干净比如在协程里写了同步的requests.get或者用了asyncio.run_until_complete那 P99 会甩出几条街。建议压测脚本里对每个查询单独统计 P50、P99、P999并且模拟真实连接池容量而不是单连接压到死。还有一个容易被忽视的点SQL 参数的传递方式影响性能。虽然上面说了缓存 key 是完整 SQL但如果你在 SQL 里把常量拼进去而不是用$1传参PostgreSQL 的 planner 会因为每次参数不同而无法复用通用计划导致执行计划频繁重新生成。反过来如果你用绑定变量plan 可以缓存。这一点在查询条件分布极不均匀的报表场景中好坏都有争议但作为默认策略我是坚定推荐绑定变量。4.4 常见报错速查表报错信息常见原因解决方案PoolAcquireTimeoutError连接池满获取连接超时调整 max_size排查泄漏降低慢查询InterfaceError: cannot perform operation: another operation is in progress同一连接上并发执行多任务事务和查询避免在同一连接上并行DataError: invalid input for query argument传了不支持的 Python 类型显式转型或注册 codecUndefinedFunctionError自定义类型/函数未注册在 init 中注册 codec或 SQL 层 castConnectionDoesNotExistError / connection is closed连接已被服务端关闭检查 max_inactive_connection_lifetime 与数据库空闲回收策略PostgresError: duplicate key value violates unique constraint唯一键冲突捕获 UniqueViolationError 处理幂等PostgresError: canceling statement due to statement timeout单条 SQL 超时优化 SQL添加索引或动态调整 command_timeout最后分享一个我自己的土办法在任何 FastAPI 中间件里对每个请求记录connection await Database.pool().acquire()的耗时如果这个耗时突然变大不用看监控你都知道池子在排队。写在后面的一点经验从最初照猫画虎把 asyncpg 塞进项目到后来能摸清它的脾气我最深的感受是asyncpg 不是那种“安上去就能跑好”的库它给了你足够的性能底子但连接池参数、事务边界、类型 codec、批量导入方式每一个都值得你提前想清楚。真把它用顺了写异步数据层的愉悦感是同步时代体会不到的。你在项目里如果也折腾过 asyncpg 或者踩过其他的坑欢迎按着这篇文章的思路自己动手验证一遍数据不会骗人跑一轮压测你就知道我说的那些细节有多重要。