本节目标：学会在异步代码里安全地使用同步库，认清几种「假异步」写法，并知道什么时候不该用 asyncio。
适用版本：Python 3.12+（实测 3.14.6）
13.3 异步生态与常见陷阱
前两节我们一直在「纯异步」的世界里写代码，但真实项目里到处都是同步库：requests、老旧的数据库驱动、某个只提供同步接口的内部 SDK。本节要解决的就是异步代码与同步世界的摩擦，以及由此衍生的一堆「看起来是异步、其实是串行」的陷阱。
13.3.1 同步库怎么用：to_thread 与 run_in_executor
13.1 节讲过：在协程里直接调同步阻塞函数，会卡死整个循环。安全的做法是把同步函数丢到线程池里执行，循环本身则继续调度别的任务。最简洁的入口是 asyncio.to_thread()：
import asyncio, time
def blocking(n):
time.sleep(0.3)
return n
async def main():
t0 = time.perf_counter()
results = await asyncio.gather(*[asyncio.to_thread(blocking, i) for i in range(3)])
print("to_thread 结果:", results, f"耗时 {time.perf_counter()-t0:.3f}s")
asyncio.run(main())
to_thread 结果: [0, 1, 2] 耗时 0.314s
三个各阻塞 0.3 秒的函数并发执行，总耗时约 0.31 秒——说明它们确实跑在多个线程里，没有互相拖累。to_thread 是 3.9+ 的语法糖，等价于 loop.run_in_executor(None, func, *args)。要控制线程池大小、或改用进程池时，才需要 run_in_executor 的完整形式（13.2.9 节）。
一句话准则：异步框架用异步库；没有异步版本时，把同步调用包进 to_thread。
13.3.2 实测：requests 在协程里会阻塞循环
「包进 to_thread」不是教条，而是有实测依据的。我们起一个本地慢接口（每个请求固定耗时 0.3 秒），用三种方式各发 5 个请求：
import time, asyncio, requests, httpx
URL = "http://127.0.0.1:8799/" # 本地服务，每个请求 0.3s
N = 5
def sync_requests():
t0 = time.perf_counter()
for _ in range(N):
requests.get(URL) # 串行阻塞
return time.perf_counter() - t0
async def requests_in_threads():
t0 = time.perf_counter()
await asyncio.gather(*[asyncio.to_thread(requests.get, URL) for _ in range(N)])
return time.perf_counter() - t0
async def async_httpx():
t0 = time.perf_counter()
async with httpx.AsyncClient(trust_env=False) as client:
await asyncio.gather(*[client.get(URL) for _ in range(N)])
return time.perf_counter() - t0
requests 串行 : 1.586s
requests + to_thread : 0.319s
httpx.AsyncClient : 0.453s
三个数字讲清了全部故事：
| 方式 | 耗时 | 原因 |
|---|---|---|
requests 串行 | 1.586s | 5 × 0.3s，一个接一个 |
requests + to_thread | 0.319s | 丢进线程池，5 个并发 |
httpx.AsyncClient | 0.453s | 原生异步，5 个并发（含建连开销） |
trust_env=False是为了忽略系统级代理设置（有些机器的代理会拦截127.0.0.1）。生产环境若不需要可省略。
再看一个更直观的对照：在协程里跑一个「心跳任务」每 0.05 秒记一次数，然后分别用 requests.get 和 httpx 请求同一个 0.3 秒的接口：
requests 直接调用: 阻塞 0.415s，期间心跳 1 次
httpx.AsyncClient: 等待 0.328s，期间心跳 8 次
requests 期间心跳只跳了 1 次——因为整条循环被冻住了；httpx 期间心跳跳了 8 次，循环照常运转。在异步代码里裸调同步 HTTP 库，等于把并发的意义全部抹掉。
13.3.3 httpx.AsyncClient：连接池与 async with
httpx 是既能同步又能异步的 HTTP 客户端。它的异步接口是 httpx.AsyncClient，有两个必须知道的点：
import httpx, asyncio
async def main():
# 用 async with 管理生命周期（trust_env=False 见 13.3.2 的说明）
async with httpx.AsyncClient(trust_env=False) as client:
r1 = await client.get("http://127.0.0.1:8799/")
r2 = await client.get("http://127.0.0.1:8799/") # 复用连接
print(r1.status_code, r2.status_code)
asyncio.run(main())
200 200
- 必须用
async with：客户端内部维护一个连接池，async with退出时统一关闭连接、释放资源。写成client = httpx.AsyncClient()而不关闭，连接会泄漏。 - 一个客户端要复用：同一个
AsyncClient里的多个请求共享连接池，第二次请求同一主机能复用已建立的 TCP 连接。这也解释了 13.3.2 里httpx那 0.45 秒里包含了首次建连的开销——预热后再发 5 个并发请求，总耗时能降到约 0.3 秒（即单个请求的时间）。
不要把 AsyncClient 写成「每次请求都新建一个」，那等于放弃了连接池。
13.3.4 异步上下文传播：contextvars 优于 threading.local
线程里保存「当前请求的上下文」常用 threading.local——每个线程一份独立的变量。但协程不是线程：成百上千个协程跑在同一个线程里，threading.local 会让它们互相覆盖。正确工具是 contextvars：
import asyncio, contextvars
request_id = contextvars.ContextVar("request_id", default="-")
async def handle(name):
request_id.set(name)
await asyncio.sleep(0.05) # 切换点：别的协程会插进来
return f"{name} -> {request_id.get()}" # 切回来仍是自己的值
async def main():
return await asyncio.gather(handle("A"), handle("B"), handle("C"))
print("contextvars:", asyncio.run(main()))
contextvars: ['A -> A', 'B -> B', 'C -> C']
三个协程在 await 处互相穿插，但每个协程读到的 request_id 都是自己设的那个值——因为 contextvars 的每个 Task 拥有独立的上下文副本，上下文在 await 切换时会自动保存与恢复。这正是日志追踪（trace id）、请求级配置需要的语义。threading.local 在同一线程的多个协程间会串味，别在异步代码里用它。
13.3.5 假异步之一：忘记 await
回到 13.1 节那个坑，它也是「假异步」的头号来源：
async def value():
return 42
async def forget():
r = value() # 忘了 await
print("忘了 await:", type(r).__name__)
asyncio.run(forget())
RuntimeWarning: coroutine 'value' was never awaited
忘了 await: coroutine
代码不报错，r 却是一个协程对象而不是 42。凡是调用 async def 函数，几乎都必须 await；唯一的例外是你故意想创建后台任务，那就该用 create_task 而不是裸调用。
13.3.6 假异步之二：循环里串行 await
即使每个操作都正确地 await 了，如果把它们顺序写出来，依然是串行：
import asyncio, time
async def task(i):
await asyncio.sleep(0.2)
return i
async def serial():
t0 = time.perf_counter()
for i in range(5):
await task(i) # 一个一个来
return time.perf_counter() - t0
async def concurrent():
t0 = time.perf_counter()
await asyncio.gather(*[task(i) for i in range(5)])
return time.perf_counter() - t0
串行 await : 1.007s
gather : 0.202s
同样的 5 个任务，串行 await 花了 1 秒，gather 只花 0.2 秒。写了 async/await 不等于并发；只要任务是彼此独立的，就应该用 gather 或 TaskGroup 把它们并发起来。判断标准：这些 await 之间有没有数据依赖？没有就该并发。
13.3.7 假异步之三：create_task 丢引用
create_task 创建的是「发射后不管」的后台任务，但如果没人持有它的引用，官方文档明确警告：任务可能在完成前被垃圾回收，从此无声消失。安全写法是把任务放进一个集合，完成时再移除：
import asyncio
background_tasks = set()
async def background(i):
await asyncio.sleep(0.3)
print(f" 任务 {i} 完成")
async def main():
for i in range(3):
t = asyncio.create_task(background(i))
background_tasks.add(t) # 持有强引用
t.add_done_callback(background_tasks.discard) # 完成后释放
await asyncio.sleep(0.5)
print(" 存活引用数:", len(background_tasks))
asyncio.run(main())
任务 0 完成
任务 1 完成
任务 2 完成
存活引用数: 0
在 3.14.6 的实测里，未保存引用的任务没有在运行中被立刻回收，而是在 asyncio.run() 结束时被统一取消——这同样是个隐患：任务「活着出去了，却没等到结果就被取消了」。无论哪种情况，结论一致：后台任务务必用集合（或变量）持有引用，并配 add_done_callback(discard) 清理。 更推荐直接用 TaskGroup，它在作用域结束前强制等待所有任务，从根上杜绝孤儿任务。
13.3.8 async for / async with / 异步生成器
异步世界里有两套特殊协议。异步上下文管理器实现 __aenter__ / __aexit__，配 async with；异步生成器用 async def + yield，配 async for：
import asyncio
class Conn:
async def __aenter__(self):
print(" 打开连接")
return self
async def __aexit__(self, *exc):
print(" 关闭连接")
async def countdown(n):
for i in range(n, 0, -1):
await asyncio.sleep(0.01)
yield i
async def main():
async with Conn():
print(" 使用连接")
print("async for:", [x async for x in countdown(3)])
asyncio.run(main())
打开连接
使用连接
关闭连接
async for: [3, 2, 1]
httpx.AsyncClient、asyncpg 的连接、各种异步文件对象都实现了 __aenter__/__aexit__，所以它们都要求 async with。异步生成器则适合「分页拉取」这类需要边等边产出的场景。
13.3.9 anyio 与 trio：另一条路线
asyncio 是标准库方案，但并非唯一。trio 从零设计，主打结构化并发（用 nursery 严格管理任务树）；anyio 是兼容层，同一份代码可跑在 asyncio 或 trio 后端上：
import anyio
async def task(name):
await anyio.sleep(0.1)
print(f" {name} 完成")
async def main():
async with anyio.create_task_group() as tg:
tg.start_soon(task, "A")
tg.start_soon(task, "B")
anyio.run(main, backend="asyncio") # 换成 backend="trio" 即可切后端
A 完成
B 完成
anyio（实测 4.15.1）的 create_task_group() 在 asyncio 后端上也提供「一个失败、全体取消」的语义。何时选哪条路线，以及 trio 的取消作用域、httpx 对双后端的支持，可以参考专题 Python 高级异步编程
。初学阶段先用标准库 asyncio 打牢基础即可。
13.3.10 什么时候不该用异步
异步不是万能药，用错场景反而更慢更复杂：
| 场景 | 该不该用异步 | 说明 |
|---|---|---|
| 大量网络 / 磁盘 I/O 并发 | 该用 | 等待期间能调度别的任务 |
| CPU 密集计算（图像、数值） | 不该 | 协程无法绕过 GIL，用多进程（12.2 节） |
| 只有一两个串行请求 | 不该 | 异步的调度开销不值当，同步代码更简单 |
| 全站都是同步库且无法替换 | 慎重 | 到处 to_thread 不如直接用线程池 |
| 需要极低延迟的少量连接 | 看情况 | 简单同步 + 线程池往往够用 |
一句话：异步解决的是「大量 I/O 等待」，不是「算得快」。 如果瓶颈是 CPU，回到第 12 章的 multiprocessing；如果根本没有并发需求，同步代码永远是最省心的选择。想更系统地看 asyncio 的实战用法，可以读专题 asyncio 完全指南
。
小结
- 同步库在协程里会卡死循环（实测
requests期间心跳仅 1 次）；用asyncio.to_thread或run_in_executor把它送进线程池（实测 5 个请求从 1.586s 降到 0.319s）。 httpx.AsyncClient要用async with管理生命周期、并复用同一个客户端以利用连接池。- 异步上下文用
contextvars而非threading.local；Task之间互不串味。 - 三种「假异步」：忘记
await、循环里串行await（1.007s vs 0.202s）、create_task丢引用。 async with/async for/ 异步生成器是异步生态的通用协议；anyio/trio是另一条结构化并发路线。- CPU 密集、无并发需求时不该用异步——异步只解决 I/O 等待。
第 12 章的线程、进程与本节的协程，构成了 Python 并发的完整版图：进程绕开 GIL 做并行计算，线程处理少量阻塞调用，协程撑起海量 I/O 并发。接下来我们要给这些代码上「保险」——下一章进入测试，学习如何用 pytest 验证你写的一切。
阅读导航：上一节：asyncio 任务、并发与超时取消 · 下一节：pytest 基础与断言 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。