asyncpg = asynchronous + PostgreSQL

深入理解 asyncpg：高性能 PostgreSQL 异步驱动实践指南

在现代微服务架构中，数据库访问逐渐从“阻塞式”转向“异步非阻塞式”。尤其是在 Python 生态中，随着 asyncio 的成熟，异步数据库驱动成为高并发场景的首选。

本文将带你深入理解 asyncpg —— 一个专为 PostgreSQL 设计的高性能异步驱动。

---

一、什么是 asyncpg？

asyncpg 是一个基于 Python asyncio 的 PostgreSQL 数据库驱动，具备以下特点：

• 🚀 极致性能（通常比 psycopg2 快 2~5 倍）
• ⚡ 原生异步（非线程池模拟）
• 🧠 高效协议实现（直接实现 PostgreSQL 二进制协议）
• 🔒 类型安全（自动映射 PostgreSQL 类型）

它不是简单的 ORM，而是一个更底层、更高效的数据库访问工具。

---

二、为什么选择 asyncpg？

在传统同步模型中：

# 同步阻塞
result = cursor.execute("SELECT * FROM users")

每一个数据库请求都会阻塞线程。

而 asyncpg：

# 异步非阻塞
rows = await conn.fetch("SELECT * FROM users")

优势非常明显：

特性	同步驱动	asyncpg
并发能力	低	高
延迟	高	低
资源利用	低	高
适合场景	简单系统	高并发服务

---

三、安装与基础使用

1. 安装

pip install asyncpg

---

2. 基础示例

import asyncio
import asyncpg

async def main():
    conn = await asyncpg.connect(
        user='user',
        password='password',
        database='testdb',
        host='127.0.0.1'
    )

    rows = await conn.fetch("SELECT id, name FROM users")

    for row in rows:
        print(row['id'], row['name'])

    await conn.close()

asyncio.run(main())

---

四、核心 API 解析

1. 查询方法

await conn.fetch(...)      # 返回多行
await conn.fetchrow(...)   # 返回单行
await conn.fetchval(...)   # 返回单个值

示例：

user = await conn.fetchrow(
    "SELECT * FROM users WHERE id=$1", 1
)

👉 注意：参数使用 $1 占位符，而不是 %s

---

2. 执行写操作

await conn.execute(
    "INSERT INTO users(name) VALUES($1)", "Alice"
)

---

3. 批量操作

await conn.executemany(
    "INSERT INTO users(name) VALUES($1)",
    [("Alice",), ("Bob",)]
)

---

五、连接池（生产必备）

在实际项目中，绝对不要频繁创建连接，应使用连接池。

示例：

pool = await asyncpg.create_pool(
    user='user',
    password='password',
    database='testdb',
    host='127.0.0.1',
    min_size=5,
    max_size=20
)

async with pool.acquire() as conn:
    rows = await conn.fetch("SELECT * FROM users")

---

为什么必须用连接池？

• 避免连接创建开销
• 控制数据库连接数
• 提升整体吞吐量

---

六、事务管理

async with conn.transaction():
    await conn.execute("INSERT INTO users(name) VALUES($1)", "Alice")
    await conn.execute("INSERT INTO logs(action) VALUES($1)", "create_user")

如果中间发生异常，会自动 rollback。

---

七、与 SQLAlchemy 2.x 的结合

在现代 Python 项目中，推荐：

asyncpg + SQLAlchemy 2.x async

示例：

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/testdb"
)

优点：

• ORM + 高性能驱动结合
• 统一数据库抽象层
• 更易维护大型项目

---

八、结合微服务架构的最佳实践

1️⃣ 每个服务只访问自己的 schema

✔ asyncpg 支持指定 schema：

SET search_path TO my_service_schema;

---

2️⃣ 禁止跨 schema 联表

❌ 错误做法：

SELECT * FROM service_a.users
JOIN service_b.orders

✔ 正确做法：

• 通过 API 获取数据
• 服务解耦

---

3️⃣ 使用连接池 + 异步模型

推荐架构：

FastAPI / Aiohttp
    ↓
asyncpg Pool
    ↓
PostgreSQL

---

九、性能优化技巧

✅ 使用 prepared statements（自动缓存）

asyncpg 默认会缓存查询计划：

await conn.fetch("SELECT * FROM users WHERE id=$1", 1)

---

✅ 避免 N+1 查询

❌：

for user in users:
    await conn.fetch(...)

✔：

SELECT * FROM users WHERE id = ANY($1)

---

✅ 使用 fetchval 提升性能

count = await conn.fetchval("SELECT COUNT(*) FROM users")

---

十、常见坑

⚠️ 1. 忘记 await

conn.fetch(...)  # ❌
await conn.fetch(...)  # ✅

---

⚠️ 2. 连接泄漏

conn = await pool.acquire()
# 忘记释放 ❌

✔：

async with pool.acquire() as conn:
    ...

---

⚠️ 3. 阻塞代码混入 async

time.sleep(1)  # ❌
await asyncio.sleep(1)  # ✅

---

十一、适用场景总结

适合使用 asyncpg 的场景：

• 高并发 API 服务（如 FastAPI）
• 微服务架构
• 实时系统（如消息处理）
• 数据密集型应用

不适合：

• 简单脚本
• 低并发后台任务

---

十二、总结

asyncpg 的核心价值可以归纳为三点：

🚀 高性能
⚡ 真异步
🧩 适配现代架构

如果你正在构建：

• 微服务系统
• 高并发 API
• 云原生应用

那么 asyncpg 基本是 PostgreSQL 的“默认推荐方案”。