协程切换开销主要来自栈帧保存/恢复、内存分配（如__await__返回的迭代器）及调度延迟；高频小任务中尤为明显，但await已完成Task或Future时走快速路径，不实际挂起。
协程切换开销主要来自哪里
Python 的 await 不是免费的——每次遇到 await，事件循环就要保存当前协程的栈帧、调度下一个就绪协程、之后再恢复。这个过程本身不耗 CPU，但会带来内存分配（如 __await__ 返回的迭代器）、上下文切换和调度延迟。尤其在高频小任务（比如每毫秒 await 一次 Redis 查询）中，开销会明显拖慢吞吐。
关键点在于：await 触发切换的条件是「表达式返回一个 awaitable」，而**不是「写了 await 就一定切」**。如果 await 的对象是已完成的 Task 或已 resolve 的 Future，CPython 会走快速路径（PyCoro_Await 内部判断），避免实际挂起。
所以减少切换的核心思路不是“少写 await”，而是：让 await 尽量落在真正需要等待的地方，并确保上游尽可能提前完成。
哪些 await 实际不触发切换（但你可能没意识到）
常见误判是认为“只要写了 await 就切”。其实以下情况通常不会产生调度开销：
-
await asyncio.sleep(0)：强制让出控制权，必然切换——这是特例，别滥用
-
await an_already_done_task：比如 task = asyncio.create_task(coro()); await task，若 task 已完成，await 几乎无开销
-
await asyncio.gather(a, b, c)：内部用 ensure_future 批量启动，await 本身只等结果，不逐个 await
-
await some_coroutine_that_returns_cached_result()：比如带 @lru_cache 的协程包装器（需手动实现，标准库不支持）
await asyncio.sleep(0)：强制让出控制权，必然切换——这是特例，别滥用await an_already_done_task：比如 task = asyncio.create_task(coro()); await task，若 task 已完成，await 几乎无开销await asyncio.gather(a, b, c)：内部用 ensure_future 批量启动，await 本身只等结果，不逐个 awaitawait some_coroutine_that_returns_cached_result()：比如带 @lru_cache 的协程包装器（需手动实现，标准库不支持）注意：asyncio.gather 和 asyncio.wait 是批量 await 的正确姿势；而 await a; await b; await c 是串行等待，不仅逻辑上慢，还多出两次调度决策成本。
await 放错位置的典型场景
最常踩坑的是在「本可并行却串行 await」和「过早 await 中间结果」：
- 把数据库查询 + 缓存检查写成
cache = await get_from_cache(); if not cache: data = await db_query() —— 应该先并发发起两者：done, pending = await asyncio.wait([get_from_cache(), db_query()], return_when=asyncio.FIRST_COMPLETED)
- 对每个 item 单独 await：
for item in items: await process(item) —— 若 process 之间无依赖，改用 await asyncio.gather(*[process(i) for i in items])
- 在日志或类型检查后立即 await：
if isinstance(x, Awaitable): result = await x —— 这种运行时判断本身就有开销，且掩盖了类型契约；应靠类型注解 + mypy 约束，而非动态分支
cache = await get_from_cache(); if not cache: data = await db_query() —— 应该先并发发起两者：done, pending = await asyncio.wait([get_from_cache(), db_query()], return_when=asyncio.FIRST_COMPLETED)
for item in items: await process(item) —— 若 process 之间无依赖，改用 await asyncio.gather(*[process(i) for i in items])
if isinstance(x, Awaitable): result = await x —— 这种运行时判断本身就有开销，且掩盖了类型契约；应靠类型注解 + mypy 约束，而非动态分支另一个隐形陷阱：await 后面跟的是同步函数调用（比如 await time.time()），这会报 TypeError: object float can't be used in 'await' expression，但如果你封装了错误的 wrapper（如 async def now(): return time.time()），表面能跑，实则白加一层协程开销。
怎么验证 await 是否真造成了瓶颈
不要猜。用 asyncio.get_event_loop().set_debug(True) 开启调试模式，它会在每次协程挂起/恢复时打日志（含耗时）。更准的方法是：
- 用
tracemalloc + asyncio.run 的 debug=True 参数，看是否频繁分配 coroutine 或 Future 对象
- 用
aiomonitor 或 py-spy record -r --pid $PID 抓取火焰图，观察 PyCoro_Await 和 handle_events 占比
- 对比两种写法的
timeit：比如 await gather(...) vs 串行 await，差距常达 2–5 倍
tracemalloc + asyncio.run 的 debug=True 参数，看是否频繁分配 coroutine 或 Future 对象aiomonitor 或 py-spy record -r --pid $PID 抓取火焰图，观察 PyCoro_Await 和 handle_events 占比timeit：比如 await gather(...) vs 串行 await，差距常达 2–5 倍特别提醒：CPython 3.12+ 引入了 “zero-cost” await 优化（针对已结束的 awaitable），但前提是你的 awaitable 确实已完成——如果逻辑里大量 await 还没 ready 的 Task，再新的解释器也救不了。
立即学习“Python免费学习笔记（深入）”；
