对账后台那个页面，打开要等 40 秒。原因很简单：200 笔订单在支付网关侧的状态是一笔一笔查回来的，网关只提供单笔查询接口，单次 P50 大约 200ms。这篇文章讲的是把它改成 asyncio 并发版本（同机、同网关，0.9 秒）的过程，以及改造过程中踩到的三个坑——它们有个共同点：代码不报错，只是慢，或者悄悄少返回几条数据。
场景：200 笔订单，串行 40 秒
# sync_version.py —— 能跑，但慢
import time
import requests
URL = "http://127.0.0.1:8080/order/{}"
def fetch_status(oid):
# 同步阻塞：线程停在 recv 上，直到网关返回
r = requests.get(URL.format(oid), timeout=3)
return r.json()["status"]
if __name__ == "__main__":
t = time.perf_counter()
res = {oid: fetch_status(oid) for oid in range(1, 201)}
print(f"{len(res)} 条，耗时 {time.perf_counter() - t:.1f}s")
200 乘 200ms 就是 40 秒，前端早就 504 了。加线程池是最省事的解法，50 个线程差不多 0.9 秒也能跑完，代价是线程创建切换的开销，以及请求量涨到几千时线程数就没法看了。asyncio 走的是另一条路：单线程，靠「等 IO 的时候切走」来并发。
为了让下面的代码都能直接跑，先把假网关起起来，每个请求延迟 200ms：
# mock_gateway.py 需要 pip install aiohttp
import asyncio
from aiohttp import web
async def handler(request):
await asyncio.sleep(0.2) # 模拟网关 200ms 的处理耗时
return web.json_response({"oid": request.match_info["oid"], "status": "paid"})
app = web.Application()
app.router.add_get("/order/{oid}", handler)
web.run_app(app, host="127.0.0.1", port=8080)
await 不产生新线程，它只是把控制权交回去
很多人第一次写 asyncio 会默认「async def 的函数就是并行的」，然后写出这种代码：
import asyncio
import time
START = time.perf_counter()
async def ticker():
for i in range(5):
await asyncio.sleep(0.1)
print(f"tick {i} at {time.perf_counter() - START:.2f}s")
async def busy():
# 纯 CPU 忙等 1 秒，中间没有任何 await
t0 = time.perf_counter()
while time.perf_counter() - t0 < 1.0:
pass
print(f"busy done at {time.perf_counter() - START:.2f}s")
async def main():
await asyncio.gather(ticker(), busy())
asyncio.run(main())
实际输出是这样：
busy done at 1.00s
tick 0 at 1.00s
tick 1 at 1.00s
tick 2 at 1.00s
tick 3 at 1.00s
tick 4 at 1.00s
5 个 tick 全挤在 1 秒之后，而且彼此间隔接近 0。因为 busy() 跑起来之后一直没把控制权交回去，事件循环轮不到处理 ticker 的定时器；等 busy 返回，5 个定时器都已经过期了，于是一次性连续触发。
这就是 asyncio 的核心约束：一个事件循环跑在一个线程里，协程之间是协作式调度，只有遇到 await 才会切换。
三个概念的关系理清楚：
| 名字 | 是什么 | 谁来推进它 |
|---|---|---|
| coroutine | async def 调用后返回的对象，函数体一行都没执行 |
没人 await 它就永远不执行，只会收到 coroutine was never awaited 警告 |
| Task | 把 coroutine 包起来挂到事件循环上，可调度、可取消 | 事件循环 |
| Future | 一个「结果还没到」的占位符，有结果时唤醒挂在上面的回调 | 事件循环调 set_result |
await fut 干的事：把当前协程剩下的部分注册成 fut 的 done callback，然后 return 回事件循环。循环拿到控制权后去 selectors（Linux 上是 epoll）上看哪些 fd 就绪、定时器堆里哪个到期，处理完再把控制权交给某个 Task。await 点是唯一的切换点，一段没有 await 的代码在整个事件循环里是原子的，不会被别的协程打断。
版本差异顺手记一下，网上很多老教程还在用淘汰写法：
- 3.7+ 用
asyncio.run(main())，不用再手写new_event_loop/run_forever/close - 3.11+ 有
asyncio.TaskGroup和asyncio.timeout()，异常和超时语义比 gather + wait_for 清楚
改成异步：信号量、连接池、超时三件套
# async_version.py 需要 pip install aiohttp
import asyncio
import time
import aiohttp
URL = "http://127.0.0.1:8080/order/{}"
async def fetch(session, sem, oid):
# 整个请求（含读 body）都包在信号量里，别只包住 session.get
async with sem:
timeout = aiohttp.ClientTimeout(total=3, connect=1)
async with session.get(URL.format(oid), timeout=timeout) as resp:
data = await resp.json()
return oid, data["status"]
async def main(ids):
sem = asyncio.Semaphore(50)
# 连接池上限要和信号量对齐，否则真正的并发上限是两者里小的那个
connector = aiohttp.TCPConnector(limit=50, limit_per_host=50)
async with aiohttp.ClientSession(connector=connector) as session:
results = await asyncio.gather(
*(fetch(session, sem, oid) for oid in ids),
return_exceptions=True, # 单笔失败不让整批结果作废
)
ok, bad = {}, []
for oid, r in zip(ids, results):
if isinstance(r, BaseException):
bad.append((oid, r))
else:
ok[r[0]] = r[1]
return ok, bad
if __name__ == "__main__":
ids = list(range(1, 201))
t = time.perf_counter()
ok, bad = asyncio.run(main(ids))
print(f"成功 {len(ok)}，失败 {len(bad)}，耗时 {time.perf_counter() - t:.2f}s")
for oid, e in bad[:3]:
print(f" {oid}: {type(e).__name__}: {e}")
对着 mock 网关跑，输出 成功 200，失败 0，耗时 0.92s：200 个请求、并发 50，分 4 批，每批 0.2 秒，再加上调度和 JSON 解析的开销。
这几个参数不是拍脑袋填的：
Semaphore(50)来自下游的承受能力。网关给你的配额是 100 QPS、单请求 200ms，那同时打过去 20 到 50 个是安全的。具体取多少要拿压测数据说话，别直接抄 50。TCPConnector(limit=...)是连接池上限，它和 Semaphore 是两把不同的锁。limit 比 Semaphore 小，任务会卡在等连接上，日志里看就是信号量早就放行了但请求发不出去；limit 比 Semaphore 大，会攒一堆空闲连接。ClientTimeout(total=3, connect=1)里的 total 从「申请连接」就开始计时，连接池排队的时间也算在内，所以它顺便保护了池满导致的长时间挂起。- 别给每笔请求都新建
ClientSession。Session 持有连接池，复用才能省掉握手；每笔都新建的话 200 个请求就是 200 次 TCP 连接（HTTPS 还要 200 次 TLS 握手），实际耗时很可能比串行还难看，退出时还会甩给你一堆Unclosed client session。
坑一：协程里混进一行同步调用，并发就白写了
改造第一版，我把大部分请求换成了 aiohttp，但有个接口留着同步请求：
# 错误写法：函数是 async 的，内部却是阻塞调用
import requests
async def fetch(oid):
await asyncio.sleep(0) # 看起来很像异步函数
r = requests.get(URL.format(oid), timeout=3) # ← 事件循环被钉死 200ms
return oid, r.json()["status"]
结果：50 并发跑 200 个请求，耗时 40 秒，跟串行一模一样，日志里没有任何异常，就是慢。
原因就是上一节说的：requests.get 在 socket recv 上阻塞的是当前线程，而事件循环也在这个线程里。它睡 200ms，整个循环就停 200ms，期间所有协程排队等着。time.sleep、读大文件、json.loads 一个几十 MB 的响应、同步 ORM 查询、往很慢的终端 print，都是同一类问题。
修法有两条。库有异步版本就直接换；没有就丢线程池：
import asyncio
import requests
from concurrent.futures import ThreadPoolExecutor
# to_thread 用的是事件循环的默认 executor，默认 max_workers = min(32, cpu+4)
# 想把并发开到 50，得自己建一个线程池
EXECUTOR = ThreadPoolExecutor(max_workers=50, thread_name_prefix="gw")
def _sync_get(oid):
r = requests.get(URL.format(oid), timeout=3)
return oid, r.json()["status"]
async def fetch_blocking(oid):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(EXECUTOR, _sync_get, oid)
# 3.9+ 更短的写法：return await asyncio.to_thread(_sync_get, oid)
# 但它走默认 executor，并发上限默认只有 32
代价得说清楚：线程不免费。Linux 默认每个线程 8MB 虚拟栈（实际 RSS 通常只有几十 KB），50 个线程还行，500 个线程时创建和上下文切换的开销就开始显形；而且 GIL 还在，_sync_get 里的 CPU 部分（比如 JSON 解析）依然串行。这条路适合「下游只提供同步 SDK」的场景，不适合当万金油。
还有个隐蔽的阻塞源：DNS。aiohttp 默认用 ThreadedResolver，把 getaddrinfo 丢进线程池，所以解析本身不阻塞循环，但会占 executor 的线程。你要是把 executor 调得很小（比如 4 个线程），几十个并发请求会堵在域名解析上，表现出来像是「连接池不够用」，方向就查歪了。
坑二：不限并发，第一个撑不住的是你自己
把信号量去掉，代码确实更短：
# 错误写法：200 个先忍了，压测时改成 5000 个
results = await asyncio.gather(*(fetch(session, oid) for oid in ids))
5000 个并发打出去，按出现顺序会发生三件事：
- 本机先炸。每个连接占一个 fd，
ulimit -n默认 1024，很快就OSError: [Errno 24] Too many open files，或者 aiohttp 抛ClientConnectorError: Cannot connect to host ...。 - 下游跟着炸。网关只有 20 个 worker，瞬时几千个连接进来，队列拍满，P99 从 200ms 涨到几秒，然后触发限流返回 429 或者直接重置连接，
bad列表一下子就长了。 - 内存往上抬。每个 Task、每个连接的读缓冲、每个没读完的响应体都在内存里，几千并发时 RSS 涨几十上百 MB 很常见。
正确的做法是让信号量覆盖整个请求生命周期，包括读 body：
# 错误写法二：信号量只包住了发请求，没包住读 body
async def fetch(session, sem, oid):
async with sem:
resp = await session.get(URL.format(oid))
# 这里 sem 已经释放，但连接还没归还连接池（body 没读完）
return await resp.json()
这段的迷惑性在于「看起来限制了并发」。实际上 sem 只限制了同时在发请求的数量，连接被占用的时间并没有变短，读 body 阶段还和别的任务交错，真正的并发上限变成了 connector.limit，信号量形同虚设；更糟的是没写 async with resp，一旦 json() 抛异常，连接不会归还，池子慢慢耗干，最后所有请求卡在 acquire 上，表现为「程序不动了也不报错」。
关于池子耗尽还有个边界：连接池等待受不受 ClientTimeout(total=...) 保护？受。total 从 acquire 开始计时，所以池满时你会拿到超时异常而不是无限挂起。但如果你用的是裸 asyncio.wait_for(session.get(...)) 又没设 ClientTimeout，那就真可能挂很久。
坑三：任务丢了、异常吞了、取消被吃掉
这三个都很短，但都属于「不报错、结果不对」那一类。
create_task 的返回值不能丢
# 错误写法：Task 对象没有任何引用
for oid in ids:
asyncio.create_task(fetch(session, sem, oid))
await asyncio.sleep(10) # 指望它们跑完
事件循环对 Task 只持弱引用。上面这段在 GC 触发时，没跑完的 Task 可能被回收，协程执行到一半就没了，日志里可能只留一句 Task was destroyed but it is pending!，也可能什么都没有。官方文档专门写了要保存引用。
tasks = [asyncio.create_task(fetch(session, sem, oid)) for oid in ids]
results = await asyncio.gather(*tasks) # 引用留在 tasks 里，顺便等结果
如果确实要「发了就不管」，至少放进集合，跑完再摘掉：
background = set()
task = asyncio.create_task(fetch(session, sem, oid))
background.add(task)
task.add_done_callback(background.discard) # 回调里顺便取异常，否则会有 Task exception was never retrieved
gather 默认不取消兄弟任务
asyncio.gather(a, b) 在默认 return_exceptions=False 下，第一个任务抛异常会立刻抛给调用方，但其他任务不会被取消，它们继续在后台跑。你在 except 里把 session 关了，后台任务就会撞上 Session is closed，或者泄漏成 Task was destroyed。
# 一个 404，整批 200 条结果全丢，剩下的还在后台跑
await asyncio.gather(*(fetch(session, sem, oid) for oid in ids))
# 至少把能拿到的拿回来
results = await asyncio.gather(*(fetch(session, sem, oid) for oid in ids), return_exceptions=True)
3.11+ 的 TaskGroup 语义正相反：任一任务失败就取消所有兄弟任务，然后把异常包成 ExceptionGroup 抛出来。
async def main(ids):
results = {}
async with asyncio.TaskGroup() as tg:
tasks = {oid: tg.create_task(fetch(session, sem, oid)) for oid in ids}
# 走到这里说明全部成功
for oid, t in tasks.items():
results[oid] = t.result()
return results
选哪个看业务：对账要「尽量多拿结果」，用 gather + return_exceptions；一批写操作要么全成要么全不成，用 TaskGroup。注意 TaskGroup 抛的是 ExceptionGroup，得用 except* 接（3.11+），except Exception 接不住。
CancelledError 不能吞
3.8 起 asyncio.CancelledError 继承 BaseException，except Exception 抓不到它——这是好事。但下面这种写法会把它吃掉：
# 错误写法
try:
data = await resp.json()
except BaseException: # 或者裸 except:
return None # 任务被取消了，却装作成功返回
后果是外层 wait_for 或 TaskGroup 的取消永远等不到完成，超时控制失效，程序在关闭时挂住。正确姿势是显式放行：
try:
data = await resp.json()
except asyncio.CancelledError:
raise # 必须重抛，这是取消协议的一部分
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
log.warning("fetch failed: %s", e)
return None
顺带区分一下 asyncio.wait_for(coro, 3) 和 aiohttp.ClientTimeout：前者在外面套一层计时器，超时后 cancel 内部任务；后者是请求层自己计时，连接池排队时间也算进去。两个都设的时候，先到的那个生效，抛出的异常类型还不一样（asyncio.TimeoutError 和 aiohttp.ServerTimeoutError），日志里要能分辨，不然排障时容易误判。
用 debug 模式把这些坑照出来
上面这些坑，静态看代码不一定看得出来，但 asyncio 自带探测器：
# 只开 asyncio 的调试模式
PYTHONASYNCIODEBUG=1 python async_version.py
# 开发模式：顺带开 warnings、资源警告和 -X dev 的其他检查
python -X dev async_version.py
debug 模式下，任何回调（包括 Task 的一步执行）耗时超过 loop.slow_callback_duration（默认 0.1 秒）都会打日志：
Executing <Task pending name='Task-7' coro=<fetch() running at async_version.py:21>
wait_for=<Future pending cb=[Task.task_wakeup()]>> took 0.202 seconds
看到这种行，基本可以确定那个位置有阻塞调用。阈值能自己调：
loop = asyncio.get_running_loop()
loop.slow_callback_duration = 0.05 # 50ms 以上就报，本地调试时用
几种常见日志对应的病因：
| 日志 | 病因 |
|---|---|
coroutine 'fetch' was never awaited |
调了 async 函数忘了 await，或者 gather 里传的是函数对象而不是调用结果 |
Task was destroyed but it is pending! |
Task 既没 await 也没取消，通常是 create_task 之后没人管 |
Unclosed client session / Unclosed connector |
ClientSession 没放进 async with，或者异常路径漏了清理 |
Task exception was never retrieved |
后台任务的异常没人取，加 add_done_callback 或改成 gather |
什么时候不该上 asyncio
写到这里得说反面。asyncio 不是「更快的 requests」，它有明确的适用边界：
- CPU 密集。GIL 摆在那，在事件循环里做计算只会把整个循环卡住，就是开篇 busy() 那个例子。该用
ProcessPoolExecutor或者干脆丢给专门的 worker 进程。 - 只调一两个下游。并发度是 2，省下来的时间还不够你调栈，
requests加个timeout=3就完事。 - 依赖链全是同步库。老版本云 SDK、没做 async 支持的 ORM、boto3（官方至今没有 async 版）。全塞
to_thread之后，代码复杂度上去了，收益和直接用ThreadPoolExecutor差不多。 - 团队没人看得懂这套栈。asyncio 的报错（
ExceptionGroup、Task exception was never retrieved、取消传播）比同步代码难查一个量级，线上排障成本得算进设计决策里。 - 需要优先级调度。事件循环的 ready 队列是 FIFO，没有优先级概念，高优任务只能靠拆队列或者限流间接实现。
还有一个一定会撞上的边界：asyncio.run() 不能嵌套，也不能在有事件循环的线程里调用。Jupyter、FastAPI 的请求处理函数、某些 Celery 配置里都会遇到：
# RuntimeError: asyncio.run() cannot be called from a running event loop
# asyncio.run(main())
await main() # 已经在循环里了，直接 await
Jupyter 顶层可以直接 await（IPython 7+）。如果在同步函数里被调用，就用 asyncio.run_coroutine_threadsafe(coro, loop) 把协程丢到那个循环所在的线程去跑，别自己再 new_event_loop()——两个循环抢同一个连接池，症状会非常难查。