📚 课程目录第4章异步I/O实战

4. 异步I/O实战

异步服务的性能上限，往往不取决于业务逻辑本身，而取决于它如何与外部世界打交道：HTTP 调用、磁盘读写、数据库查询。这三类 I/O 是异步编程收益最明显、也最容易踩坑的地方。本节围绕这三条主线展开，先讲客户端选型与用法，再讲文件操作的边界，最后落到数据库驱动与连接池的工程决策。

一、HTTP 客户端：aiohttp 与 httpx 的分工

aiohttp 是纯异步栈的老牌方案，客户端与服务端一体，性能好、生态成熟；httpx 则提供同步与异步双接口，API 更接近 requests，对迁移友好，且原生支持 HTTP/2。选型时可以先问三个问题：是否需要 HTTP/2、是否要复用同步代码、是否依赖 aiohttp 特有的服务端能力。

两者最关键的共同点是：连接必须复用，客户端必须长生命周期。每次请求都新建 ClientSession 或 AsyncClient，会反复做 TCP 握手与 TLS 协商，异步带来的并发优势会被连接开销吃掉。正确做法是在应用启动时创建一个客户端实例，通过依赖注入或全局单例共享，在关闭时统一释放。

import aiohttp, asyncio

async def fetch_all(urls):
    connector = aiohttp.TCPConnector(limit=100, limit_per_host=20)
    timeout = aiohttp.ClientTimeout(total=5, connect=1)
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        tasks = [fetch_one(session, u) for u in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

async def fetch_one(session, url):
    async with session.get(url) as resp:
        resp.raise_for_status()
        return await resp.json()

这里 limit=100 是全局并发连接上限，limit_per_host=20 限制对单个域名的连接数，避免把下游打挂。ClientTimeout 把总超时与连接超时分开设置，比单一超时更可控。httpx 的写法几乎对称：httpx.AsyncClient(limits=httpx.Limits(max_connections=100, max_keepalive_connections=20), timeout=httpx.Timeout(5.0, connect=1.0))。

并发请求务必用 asyncio.gather(..., return_exceptions=True)，否则单个请求抛异常会取消整批任务，且异常信息难以定位到具体 URL。

二、异步文件操作：没有真正的“异步磁盘”

这是最容易被误解的一点：操作系统层面并没有通用的异步文件读写接口，aiofiles 的本质是把阻塞的文件操作丢进线程池执行。因此它的价值在于不阻塞事件循环，而不是让磁盘变快。对于小文件、低频读写，直接同步读反而更简单；对于大文件或高频日志写入，才值得引入 aiofiles。

import aiofiles

async def write_log(path, lines):
    async with aiofiles.open(path, "a", encoding="utf-8") as f:
        for line in lines:
            await f.write(line + "\n")

需要注意：aiofiles.open 返回的上下文管理器必须在 async with 中使用，逐行 await f.write() 会产生多次线程切换，批量写入时应先拼接成较大块再一次性写出。若追求极致吞吐，可考虑 loop.run_in_executor 配合普通文件对象，控制粒度更灵活。

三、异步数据库驱动选型

PostgreSQL 场景下，asyncpg 是性能标杆，协议层直连、参数绑定高效，但不兼容 DB-API，SQL 需自行管理；SQLAlchemy 2.0 的 async 引擎则提供 ORM 与连接池抽象，适合中大型项目。MySQL 侧常用 aiomysql 或 asyncmy，后者性能更好但生态略小。选型建议：追求极致性能且 SQL 可控，选 asyncpg；需要 ORM、迁移与多数据库支持，选 SQLAlchemy async + asyncpg 驱动。

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:pwd@db:5432/app",
    pool_size=10, max_overflow=20,
    pool_timeout=5, pool_recycle=1800,
    echo=False,
)
Session = async_sessionmaker(engine, expire_on_commit=False)

async def get_user(uid):
    async with Session() as s:
        return await s.get(User, uid)

四、连接池参数怎么定

连接池不是越大越好。数据库能承受的并发连接有限，池过大只会让请求在数据库侧排队，甚至触发连接数上限。经验公式是：池大小 ≈ 服务实例数 × 单实例并发查询数，再留出少量 overflow 应对突发。以 4 个实例、每实例 10 个并发查询为例，单实例 pool_size=10、max_overflow=5 即可，总连接数控制在 60 以内。

参数	作用	示例值	调优方向
pool_size	常驻连接数	10	按并发查询数估算
max_overflow	突发额外连接	5	不超过数据库余量
pool_timeout	取连接等待上限	5s	过小会误报超时
pool_recycle	连接回收周期	1800s	小于数据库空闲断开时间

连接池耗尽时，请求会卡在 pool_timeout 上，表现为接口整体变慢而非报错。排查时应先看池等待指标，再决定是扩池还是优化慢查询，盲目扩池可能把压力转移到数据库。

把三条线放在一起看，异步 I/O 的工程要点是一致的：复用昂贵资源（连接、会话、文件句柄）、给等待设上限（超时、池等待）、让异常可定位（隔离、日志、指标）。HTTP 客户端管好连接池与超时，文件操作认清线程池本质并控制粒度，数据库驱动选对协议层并算准池大小，异步服务才能真正把并发能力转化为吞吐。