第 16 章异步生态实战：httpx 与异步数据库

这一章把 asyncio 落到实处：同步 requests 和异步 httpx/aiohttp 到底差在哪，异步数据库驱动怎么用，以及「同步阻塞代码混进事件循环」为什么是生产事故。最后给你一把尺子——什么时候真的值得上异步。

墨叔：老哥，上一章你把 asyncio 的引擎舱看明白了，但落地时最容易栽的坑是——生态不匹配。你在协程里兴高采烈 import requests 然后 requests.get，以为并发起来了，结果压测一上，QPS 反而掉、延迟飙红。你猜问题出在哪？

老哥：……requests 是同步阻塞的，我在事件循环里调它，把整个循环堵住了？

墨叔：正中靶心。这一章就讲「异步的生态怎么配齐」：同步 requests vs 异步 httpx/aiohttp、异步数据库驱动、同步代码混进循环的灾难，以及何时真的值得上异步的判据。记住一句总纲——asyncio 不是靠你写对 async/await 就成了，整条 IO 链路上的库都得是 async 的，有一个同步短板，木桶就漏了。

一、同步 requests vs 异步 httpx / aiohttp（怎么用）

先看清差距。同步 requests 一次请求，线程就卡在等响应上：

import requests
 
def fetch_sync(url: str) -> int:
    resp = requests.get(url, timeout=5)
    return resp.status_code
 
# 串行打 3 个接口：耗时 ≈ 三者响应时间之和
for u in urls:
    print(fetch_sync(u))

在普通「每请求一个线程」的服务里，requests 没问题——线程卡着等 IO，但别的请求有别的线程接。可一旦你进了 asyncio 单线程事件循环，requests.get 这种同步阻塞调用会卡死整个循环，循环里其它所有协程都跟着饿死。这就是上一章说的「协程里混入阻塞调用」。

异步客户端 httpx（推荐，API 和 requests 几乎一致，且同步异步双模）这么写：

import asyncio
import httpx
 
async def fetch_async(client: httpx.AsyncClient, url: str) -> int:
    resp = await client.get(url, timeout=5.0)
    return resp.status_code
 
async def main(urls: list[str]) -> None:
    # AsyncClient 内部是连接池 + 异步 IO，单线程并发扛多个请求
    async with httpx.AsyncClient() as client:
        tasks = [asyncio.create_task(fetch_async(client, u)) for u in urls]
        for coro in asyncio.as_completed(tasks):
            code = await coro
            print("完成一个:", code)
 
if __name__ == "__main__":
    asyncio.run(main(urls))

aiohttp 是另一主流异步 HTTP 库，性能略猛但 API 更底层（要自己管 session、错误处理更啰嗦）。我的建议：新项目无脑 httpx——它既能 requests 风格同步用，也能 AsyncClient 异步用，团队迁移成本低。httpx 安装：pip install httpx；aiohttp：pip install aiohttp。

老哥：那如果项目里一大堆 requests 老代码，我是不是得全改写成 httpx？

墨叔：不一定。两个解法：(1) 隔离边界——把阻塞的 requests 调用扔到线程池跑，别让它进循环：await loop.run_in_executor(None, requests.get, url)。这样老代码不动，又不堵循环。(2) 新写的部分直接 httpx 异步，逐步替换。但注意 run_in_executor 本质是「开个线程去阻塞」，并发上限受线程数约束，高并发场景还是原生异步客户端更稳。

二、异步数据库驱动（什么场景用）

HTTP 之外，数据库是服务第二大 IO 源。同步驱动（如 psycopg2、pymysql）在 asyncio 里同样会卡循环。异步驱动把「发查询、等结果」也做成非阻塞：

import asyncio
import asyncpg   # PostgreSQL 异步驱动，pip install asyncpg
 
async def query_users(ids: list[int]) -> list[tuple]:
    # 创建连接池（不是每查一次建连，那会炸）
    pool = await asyncpg.create_pool(
        user="app", password="pwd", database="shop", host="127.0.0.1",
        min_size=2, max_size=10,
    )
    async with pool.acquire() as conn:          # 从池里借一条连接
        rows = await conn.fetch("SELECT id, name FROM users WHERE id = ANY($1)", ids)
    await pool.close()
    return [(r["id"], r["name"]) for r in rows]
 
if __name__ == "__main__":
    asyncio.run(query_users([1, 2, 3]))

asyncpg 是 PostgreSQL 最快的 Python 驱动（底层用 Cython，协议级优化），支持连接池、预处理语句、监听通知。MySQL 阵营用 aiomysql（pip install aiomysql），用法类似。关键点：必须 await 借连接、必须 await 查询——漏掉 await，你拿到的是「还没执行的协程对象」而不是结果，这类 bug 编译期发现不了，运行时才炸，和 Go 的「忘了 <-ch」异曲同工，但 Go 至少有类型提醒，Python 全靠你细心。

对照 Go：你用 database/sql + pgx，db.Query 在非事务路径上默认是同步阻塞，但 goroutine 真并行，所以「阻塞」不致命；Python 异步 DB 驱动则是把查询也协程化，配合事件循环单线程并发。两者都能扛高并发，但 Python 版强制要求整条调用链 async，包括你封装的 repo 层、service 层——这就回到「函数着色」：一旦用异步 DB，上层函数全得 async，否则 await 用不了。

三、同步阻塞代码混进事件循环的灾难（收益与代价）

墨叔：老哥，我见过最惨的线上事故，就是有人在 FastAPI 的 async 路由里调了 requests.get 和 pandas.read_csv。表面跑得欢，一压测，单实例 QPS 从预期的 2000 掉到 30。你猜根因？

老哥：循环被那几个同步调用堵死了，后面的请求全排队？

墨叔：对，而且比排队更隐蔽。事件循环是单线程，任何一个协程执行「不 await 的同步重活」，这段时间循环完全停摆——所有其他协程（可能几百个都在等 IO）全被冻结。你以为在并发，实际退化成「单线程串行 + 一个长阻塞」。灾难有三层：

延迟雪崩。一个 requests.get 等 200ms，这 200ms 内循环上所有协程都动不了，它们的 IO 回调也无法被处理，累计延迟指数级放大。
吞吐假死。你开了连接池、起了几百协程，但因为循环被卡，它们根本没机会跑，资源空转。
超时连环爆。循环卡住导致 asyncio.wait_for 的闹钟也「来不及响」（循环没空处理定时器），超时机制失效，下游慢响应直接拖垮整服务。

给个能复现的「反面教材」代码，你本地跑会看到明显卡顿：

import asyncio
import time
 
def blocking_work() -> None:
    time.sleep(1)   # 同步阻塞！在协程里调用 = 卡死整个事件循环 1 秒
 
async def coro(tag: str) -> None:
    print(f"{tag} 开始")
    blocking_work()            # 灾难点：没有 await，循环在此冻结
    print(f"{tag} 结束")
 
async def main() -> None:
    await asyncio.gather(coro("A"), coro("B"), coro("C"))
 
asyncio.run(main())
# 预期 A/B/C 并发交错，实际：A 卡 1s → B 卡 1s → C 卡 1s，总 3s 串行

正确解法：要么把阻塞活儿丢线程池，要么用异步等价物。

async def coro_fixed(tag: str) -> None:
    print(f"{tag} 开始")
    # 把同步阻塞函数挪到默认线程池执行，循环不被卡
    await asyncio.get_running_loop().run_in_executor(None, time.sleep, 1)
    print(f"{tag} 结束")

run_in_executor 的语义是「借一个线程去干这个阻塞活，我（协程）先让出，干完了再回来」——这正是 Go 里 goroutine 碰到阻塞系统调用时 runtime 自动做的事，只不过 Python 要你手动声明。代价：线程池有上限（默认 ThreadPoolExecutor 按 CPU 数定大小），海量阻塞调用会耗尽线程、又开始排队。所以根治还是「IO 全用 async 库」，线程池只当兜底桥接。

四、何时真的值得上异步（判据）

老哥：那我是不是所有 IO 服务都该上 asyncio？

墨叔：千万别。上异步有成本——整条链 async 化、调试更难（堆栈跨协程）、生态得配齐、新人上手曲线陡。什么时候「真的值得」，给你一把硬尺子：

值得上异步，当且仅当满足下面多条：

并发连接数高（成百上千以上），且瓶颈在等待 IO（HTTP、DB、RPC），不是计算；
单实例要扛高 QPS，且对内存敏感（协程比线程省一个数量级内存）；
需要精细控制超时/取消/限流（asyncio 的 wait_for/cancel/Semaphore 表达力强）；
团队能接受「全栈 async」的约束，依赖库都有 async 版。

不值得上异步，以下场景退回同步 + 多线程更省心：

并发量低（每秒几十请求），用 requests + ThreadPoolExecutor 足够，改 async 纯增复杂度；
任务里 CPU 密集占比高（asyncio 帮不了，得上多进程）；
依赖的老库全是同步的，没有 async 版，强上异步得大量 run_in_executor 桥接，得不偿失；
团队对 asyncio 不熟，出协程相关 bug 排错慢——先用稳定的同步栈，比「漂亮但救不了火」重要。

给你一个决策小公式：先写同步单线程版 → profile 确认瓶颈是 IO 等待且并发高 → 再上 asyncio。不要反过来「为异步而异步」。我见过太多项目，把 CRUD 服务重写成全 async，结果 QPS 没变、bug 翻三倍——纯属自找麻烦。

最后补一个生产常用件：asyncio.Semaphore 做并发限流，避免一个协程函数瞬间打出一万个下游请求把人家的服务打挂：

import asyncio
import httpx
 
sem = asyncio.Semaphore(20)   # 最多同时 20 个下游请求
 
async def fetch_limited(client: httpx.AsyncClient, url: str) -> int:
    async with sem:           # 超过 20 并发就在这里排队
        resp = await client.get(url, timeout=5.0)
        return resp.status_code
 
async def crawl(urls: list[str]) -> list[int]:
    async with httpx.AsyncClient() as client:
        return await asyncio.gather(*(fetch_limited(client, u) for u in urls))

对照 Go，这就是 chan struct{} 当信号量或 golang.org/x/sync/semaphore——思路一致，asyncio 用上下文管理器 async with sem 表达「获取/释放」，读起来更顺。

回到开头那句话：asyncio 成败不在你会不会写 await，在整条 IO 链路的库是否都 async。requests 换 httpx、psycopg2 换 asyncpg、别在协程里留同步阻塞——这三件事做对，并发才真起来；否则就是「假异步、真串行」，压测一打就露馅。

六、再往深一层：连接池、背压与异步服务的稳定性

老哥：墨叔，我按你说的把 requests 换成 httpx、psycopg2 换成 asyncpg 了，链路全 async，可压测到高并发还是崩。还有哪漏了？

墨叔：漏的往往是「资源池」和「背压」这两块，它们和并发模型同等重要，却常被异步新手忽略。先说连接池：无论是 HTTP 的 httpx.AsyncClient 还是数据库的 asyncpg.create_pool，底层都维护一组复用连接。如果你每次请求都 new 一个客户端、查完就关，连接不复用，TCP 三次握手和 TLS 握手的代价会吞掉异步带来的所有收益，高并发下直接把下游或服务自身拖死。正确做法是整个应用生命周期共用一个客户端 / 一个连接池，像前面例子用 async with 包住批次。更进一步，连接池要设上限（httpx 的 limits、asyncpg 的 max_size），否则并发一万请求就会真去建一万个连接，把文件描述符和对方服务一起打爆。

再说背压。异步的「来者不拒」是双刃剑：事件循环来多少协程接多少，没有 OS 线程数那种天然上限，于是你一个接口被刷，瞬间起五万个协程去打下游，下游先挂、你自己的连接池也耗尽、连环雪崩。解决办法就是上一章提到的 asyncio.Semaphore 限流，以及更上层用队列做削峰（把请求塞进 asyncio.Queue，worker 协程按固定速度消费）。背压的本质是「承认自己处理能力有上限，超出的要么排队要么快速失败」，这是异步服务能不能扛压测的分水岭。

最后一句收尾：异步让你「能并发」只是第一步，能不能「稳并发」取决于连接池复用、并发限流、超时取消三件套是否齐全。很多团队 async 上线后反而更脆，根子不在 asyncio 不行，是只上了并发、没上控制。把这三件套补齐，你才真正从「会写异步」走到「能扛生产的异步」。

老哥：墨叔，我还有个现实顾虑：团队里一半人没写过 asyncio，我强推异步，他们写出 bug 排不动，是不是反而害了项目？

墨叔：你这个顾虑非常成熟，技术选型从来不是「哪个更快就选哪个」，是「哪个在团队手里最稳就选哪个」。asyncio 的学习曲线确实陡：函数着色、协程静默不执行、循环被同步调用卡死、堆栈跨协程难读——这每一处都能让不熟悉的人写出「能跑但随时崩」的代码，而这类 bug 往往在压测或半夜告警时才暴露。所以如果团队对 asyncio 普遍陌生、且业务并发量没有高到必须异步，我反而建议先用同步加线程池把功能做稳，等并发真成瓶颈、且有同事愿意啃下 asyncio 再迁移。反过来，如果你们要做的是高并发网关、爬虫、实时推送这类「天生异步」的服务，那就该早点上 asyncio，并把「全栈 async 纪律」写进代码评审清单：协程里禁写同步阻塞、连接必须池化、超时取消必须显式。一句话总结——异步是放大器：团队懂它，它放大吞吐；团队不懂它，它放大故障。选型时把「人」算进去，比把「QPS」算进去更关键。

【思考题】

你的 FastAPI 路由是 async def，里面调了老代码 pandas.read_csv("big.csv")（同步、读 2 秒）。压测时单实例吞吐暴跌。请说出两个修复方案，并比较哪个更适合「这个 CSV 读取是偶发、不是每请求都做」的场景。
项目里 psycopg2（同步 PG 驱动）用了三年，现在想上 asyncio 提高并发。列出「全量改成 asyncpg」之外，至少两种迁移路径，并说清每种路径下「混合栈」要注意什么（比如 async 路由里调同步 DB 函数）。
给你一把尺子：「并发连接数 N」和「单请求 IO 等待占比 p」。当 N 很小（如 50）时，为什么同步多线程往往比 asyncio 更划算？当 N 很大（如 5000）且 p 接近 1 时，asyncio 的优势从哪来（内存？调度？）？请结合事件循环单线程本质说明。

【参考答案】

两个方案：(a) run_in_executor 桥接——await loop.run_in_executor(None, pd.read_csv, "big.csv")，把阻塞读取丢到线程池，事件循环不被卡，其它协程照常跑。改一行、老代码不动。(b) 全改成异步 CSV 读取——用 aiofiles 异步读文件 + 流式解析，但 pandas 本身不支持真正异步，得换 polars 的懒加载或自己分块。对「偶发、不是每请求都做」的场景，(a) 更合适：改动最小、不动老逻辑，偶发调用借个线程池代价可忽略；没必要为偶发操作重构整个读取链路。注意线程池默认大小有限，若偶发变频繁且并发高，要评估线程耗尽风险，那时才考虑 (b) 或把 CSV 预加载到内存/缓存，根本不每次读盘。
两种迁移路径：(a) 边界隔离——保持 psycopg2 同步 DB 函数不动，在 async 路由里用 await loop.run_in_executor(None, sync_db_query, args) 调用，把阻塞 DB 调用挪到线程池。注意：线程池有上限，高并发下同步 DB 调用会占满线程、又开始排队，且每条连接在线程间不共享（需连接池配足够线程）。混合栈下要清楚「哪里是 async 边界、哪里是同步落点」，避免线程池耗尽。(b) 新写 async 层、旧层逐步替换——新接口用 asyncpg 写，老接口保留 psycopg2，repo 层按接口新旧分流；等老接口调用量降为零再删同步代码。注意：混合栈里同步 DB 函数和 async 函数不能简单互调，async 里调同步必须 run_in_executor，同步函数里绝不能直接 await（会报「没有运行中的循环」），需用 asyncio.run 包一层或重构为 async。核心是「async 边界清晰、同步落点用线程池兜底」。
N=50 时同步多线程更划算的原因：并发低，线程数（50）完全在系统承受范围内，每请求一线程、阻塞就等，简单可靠；上 asyncio 要全栈 async 化、调试更难，收益（省内存/高并发调度）在 50 并发下几乎体现不出来，反而增复杂度。N=5000 且 p≈1（几乎全在等 IO）时 asyncio 优势来自：(a) 内存——5000 线程每个栈 8MB 约 40GB，直接 OOM；5000 协程共享单线程栈，内存以 MB 计。(b) 调度——事件循环是单线程协作式，5000 协程在 IO 等待时通过 await 让出，循环在就绪回调间零成本切换，没有线程上下文切换开销；而 5000 线程靠 OS 抢占式调度，切换成本高、缓存命中差。(c) 结合第 14/15 章：asyncio 单线程事件循环在 IO 密集高并发下「一个厨师看千口锅」，厨师永远不闲，吞吐拉满；线程模型在同等并发下被内存和切换成本压垮。所以判据本质是「并发量 × IO 占比」决定 async 是否值回改造成本。

徐术挺-漫思录

探索

16-异步生态实战

第 16 章异步生态实战：httpx 与异步数据库

一、同步 requests vs 异步 httpx / aiohttp（怎么用）

二、异步数据库驱动（什么场景用）

三、同步阻塞代码混进事件循环的灾难（收益与代价）

四、何时真的值得上异步（判据）

六、再往深一层：连接池、背压与异步服务的稳定性

目录

徐术挺-漫思录

探索

16-异步生态实战

第 16 章 异步生态实战：httpx 与异步数据库

一、同步 requests vs 异步 httpx / aiohttp（怎么用）

二、异步数据库驱动（什么场景用）

三、同步阻塞代码混进事件循环的灾难（收益与代价）

四、何时真的值得上异步（判据）

六、再往深一层：连接池、背压与异步服务的稳定性

目录

第 16 章异步生态实战：httpx 与异步数据库