Python 异步编程入门：asyncio 的核心概念与常见陷阱

很多人第一次接触 asyncio 时的反应是：代码里加了 async 和 await，为什么跑得反而更慢了？这通常是因为只学了语法，没有理解它背后的调度模型。这篇文章从事件循环讲起，把几个最常踩的坑一次性说清楚。

并发不等于并行

先建立一个基本认知：asyncio 是单线程的并发模型。它只用一个线程、一个事件循环，靠“在某件事等待时切换去做别的事”来提升吞吐。适合的场景是 I/O 密集型任务，比如网络请求、读写文件、数据库查询，这些操作的绝大部分时间都花在等待上。

而对于 CPU 密集型任务，比如图像处理、大量数值计算，async 没有任何帮助，因为计算本身就会一直占着那唯一的线程，没有等待窗口可供切换。这类任务应该交给多进程或专门的计算型 Worker。

三个核心概念

协程是用 async def 定义的函数，调用它不会立刻执行，而是返回一个协程对象。事件循环负责调度这些协程。Task 则是把协程包装成可被并发调度的实体，只有被包装成 Task，协程才能真正“同时”跑起来。

import asyncio

async def fetch(name, delay):
    print(f"{name} 开始")
    await asyncio.sleep(delay)
    print(f"{name} 结束")
    return name

async def main():
    # 顺序执行：总耗时约 3 秒
    await fetch("A", 1)
    await fetch("B", 2)

asyncio.run(main())

把上面改成 gather，两个任务就会并发执行，总耗时变成约 2 秒，也就是两件事里较慢的那一件的耗时：

async def main():
    results = await asyncio.gather(
        fetch("A", 1),
        fetch("B", 2),
    )
    print(results)  # ['A', 'B']

asyncio.run(main())

asyncio.gather 会等待全部完成并返回结果列表，顺序与传入顺序一致。如果希望某个任务先跑起来、稍后再取结果，可以用 asyncio.create_task 拿到 Task 对象，之后再 await 它。

坑一：在协程里调用阻塞函数

这是最致命也最常见的问题。事件循环是单线程的，一旦你在协程里调用了同步的数据库驱动、requests.get、time.sleep，整个循环都会被卡住，其他所有任务一起停摆。此时并发优势荡然无存，甚至可能比纯同步代码更慢。

解决办法有两条：改用异步库，例如 httpx、aiohttp、asyncpg；或者确实无法替换时，用 await asyncio.to_thread(func, args) 把阻塞调用丢到线程池里执行，让事件循环腾出手来。

坑二：忘记 await

忘记 await 不会报错，只会得到一个未被调度的协程对象，同时冒出一句 “coroutine was never awaited” 的警告。结果就是任务静默地没执行，排查起来非常费时间，尤其是在跨函数调用的场景里。建议在开发环境把这类警告升级为异常，让问题在测试阶段就暴露出来。

坑三：取消与异常处理

异步任务被取消时，会在最近的 await 处抛出 CancelledError。如果你在协程里写了宽泛的 except Exception，要特别小心不要把它吞掉，否则任务无法正常退出，可能造成连接泄漏。正确的做法是把清理逻辑放进 finally，并让异常继续向上传递。另外，gather 默认只要有一个任务抛异常就会向外抛出，其余任务不会自动取消，需要显式处理这种半完成的中间状态。

什么时候不该用

如果你的程序本身没有并发 I/O，或者并发量很小，引入 asyncio 只会增加代码复杂度：所有调用链上的库都必须是异步的，任何一处遗漏都可能阻塞事件循环。技术选型要看收益，而不是看热度。对于只有几个接口调用的脚本，同步代码往往更清晰、更好维护，也更容易被后来的人接手。