一次性彻底搞懂异步编程：从"为什么需要"到"怎么用"到"底层原理"
先把最核心的一句话记住：async/await 是 Python 实现「协程（Coroutine）」的语法糖，用来写「异步非阻塞」代码。
在函数定义前加 async，这个函数就变成了一个协程函数。调用它不会立即执行，而是返回一个协程对象（需要事件循环来驱动执行）。
在协程函数内部，用 await 等待一个可等待对象（协程 / Task / Future）。等待期间，控制权交还给事件循环，可以去执行其他协程。
Python 3.7+ 的入口函数。创建一个事件循环，运行传入的协程，直到它完成，然后关闭事件循环。你现在只需要记住：异步代码从这里开始。
能用 await 等待的对象，包括：协程（coroutine）、Task、Future。普通函数返回值不能用 await！
下面所有代码都可以直接复制运行（Python 3.7+）。
# 定义一个协程函数（不会马上执行！） async def say_hello(): print("Hello") await asyncio.sleep(1) # 模拟IO等待1秒 print("World") return "done" # 正确调用方式：用 asyncio.run() import asyncio result = asyncio.run(say_hello()) print(result) # 输出: done
import asyncio, time # 模拟一个网络请求 async def fetch_data(name, seconds): print(f"{name} 开始请求...") await asyncio.sleep(seconds) # 模拟IO等待 print(f"{name} 请求完成！") return f"{name}的数据" async def main(): start = time.time() # ✅ 并发执行：三个请求同时发出 # gather 会同时启动所有协程，总耗时 ≈ 最慢的那个 results = await asyncio.gather( fetch_data("请求A", 2), fetch_data("请求B", 3), fetch_data("请求C", 1), ) print(f"结果: {results}") print(f"总耗时: {time.time()-start:.2f}秒") # 输出: 总耗时 ≈ 3秒（不是 2+3+1=6秒！） asyncio.run(main())
async def main(): # create_task: 把协程包装成 Task，立即加入事件循环调度 # 不像 await 会阻塞，create_task 马上返回 task1 = asyncio.create_task(fetch_data("A", 2)) task2 = asyncio.create_task("B", 3)) # 这里可以做其他事情... print("任务已提交，做点别的...") # 需要结果时再 await result1 = await task1 result2 = await task2 print(f"结果: {result1}, {result2}")
# ❌ 错误：直接调用协程函数，啥也不会发生！ say_hello() # 只是返回了一个协程对象，没有执行！ # 输出: RuntimeWarning: coroutine was never awaited # ❌ 错误：用普通方式调用（不是异步环境） result = say_hello() # 得到的是 coroutine 对象，不是结果！ # ✅ 正确：用 await 或 asyncio.run() asyncio.run(say_hello()) # 方式1：作为入口 result = await say_hello() # 方式2：在协程内部
事件循环是异步的核心引擎。下面用可视化动画展示它的工作原理。
| 对比维度 | 同步（普通函数） | 异步（async/await） | 多线程（threading） |
|---|---|---|---|
| 执行方式 | 顺序执行，一行等一行 | 协程协作式，await 时让出控制权 | 操作系统抢占式，真正并行（多核） |
| 适用场景 | CPU 密集型（计算多） | IO 密集型（网络/文件） | IO 密集型 + 部分 CPU 密集型 |
| 开销 | — | 极低（协程是用户态，切换快） | 较高（线程切换有系统开销） |
| 并发量 | 1个 | 数万个协程无压力 | 几百个线程就吃力 |
| 代码复杂度 | 最简单 | 中等（要理解 await 机制） | 高（要考虑锁、竞争条件） |
| GIL 影响 | 受 GIL 限制，多核无力 | 单线程，不受 GIL 影响（但也不能利用多核） | 受 GIL 限制（IO 时可释放 GIL） |
# ❌ 错误：time.sleep() 会阻塞整个事件循环！ async def bad(): time.sleep(1) # 整个程序卡住1秒，其他协程也无法执行！ # ✅ 正确：用 asyncio.sleep() async def good(): await asyncio.sleep(1) # 只挂起当前协程，其他协程可以继续
# ❌ 错误：创建了协程但没 await，任务永远不会执行 async def main(): asyncio.create_task(fetch_data("A",1)) # 函数结束了，但任务可能还没跑完！ # ✅ 正确：保存 task 引用并 await async def main(): task = asyncio.create_task(fetch_data("A",1)) await task # 等待任务完成
# ❌ 错误：CPU 密集计算会阻塞事件循环 async def slow(): for i in range(10**8): _ = i * i # 没有 await，事件循环无法切换！ # ✅ 正确：用 run_in_executor 把计算放到线程池 async def slow_fixed(): await asyncio.get_event_loop().run_in_executor( None, # 使用默认线程池 lambda: heavy_compute() # 放到其他线程执行 )
🎉 现在你已经理解 async/await 了！
记住：await = "我先歇会儿，你先跑"