Python
Python 异步编程：asyncio 核心机制与实践
理解协程、事件循环、await、gather 与 create_task，并掌握 asyncio 的异常处理。
TIP 核心直觉 同步 = 傻等；多线程 = OS 帮你分身；协程 = 自己主动让位
一、三种模式对比
| 同步 | 多线程 | 协程（asyncio） | |
|---|---|---|---|
| 执行方式 | 顺序，一个跑完才跑下一个 | OS 调度多线程并行 | 单线程，await 主动切换 |
| 总耗时 | 所有任务之和 | 最慢那个任务 | 最慢那个任务 |
| 适合场景 | 简单脚本 | I/O 密集，不改代码结构 | 大量并发 I/O（爬虫、API） |
| 开销 | 无 | 线程内存开销大 | 极轻量 |
WARNING GIL 陷阱 Python 多线程有全局解释器锁（GIL），同一时刻只有一个线程跑 Python 代码。 所以多线程适合等待 I/O，不适合CPU 密集计算。
二、协程语法全景
2.1 async def — 定义协程函数
async def 烧水():
return "热水"
# ⚠️ 调用后不会立刻执行！返回的是协程对象
coro = 烧水() # <coroutine object>，代码没跑NOTE 记住
async def定义的函数，调用后只是拿到一个协程对象，必须被"驱动"才真正执行。
2.2 驱动协程执行的四种方式
# 方式 1：asyncio.run() —— 最外层唯一入口
asyncio.run(烧水())
# 方式 2：await —— 在另一个协程内部调用
async def main():
result = await 烧水()
# 方式 3：asyncio.gather() —— 并发多个，等全部完成
await asyncio.gather(烧水(), 切菜())
# 方式 4：asyncio.create_task() —— 提交给事件循环，可单独控制
async def main():
task = asyncio.create_task(烧水())
await task2.3 await vs create_task — 核心区别
一句话：
await会等，create_task不等。
# await：提交 + 立刻等，期间不往下走
async def main():
result = await 烧水()
print("这行要等烧水完才执行") # 串行
# create_task：提交后立刻继续，不等
async def main():
task = asyncio.create_task(烧水())
print("这行立刻执行，烧水在后台跑") # 并发
result = await task # 这里才真正等做饭类比：
await 烧水()→ 站炉子旁边盯着，水没开不走create_task(烧水())→ 点火就去切菜，需要热水再回来
await 协程() |
create_task(协程()) |
|
|---|---|---|
| 何时开始执行 | 立刻 | 立刻（下一事件循环周期） |
| 会不会阻塞当前 | 会 | 不会 |
| 能否取消 | 不能 | 能（task.cancel()） |
| 适合场景 | 顺序依赖 | 并发、需要控制生命周期 |
2.4 并发运行：gather vs create_task
import asyncio
async def 任务(名字, 秒数):
print(f"{名字} 开始")
await asyncio.sleep(秒数)
print(f"{名字} 完成")
return f"{名字}的结果"
# gather：一批提交，等全部完成，返回结果列表
async def main():
results = await asyncio.gather(
任务("A", 3),
任务("B", 1),
任务("C", 2),
)
# results 顺序永远按传入顺序，不按完成顺序
print(results) # ['A的结果', 'B的结果', 'C的结果']
# 耗时 ~3 秒（取最长）
# create_task：提交后还能干别的，可单独操作
async def main():
task_a = asyncio.create_task(任务("A", 3))
task_b = asyncio.create_task(任务("B", 1))
print("任务已提交，我先干别的...")
await asyncio.sleep(0.5)
result_a = await task_a
result_b = await task_bTIP gather ≈ create_task + await 的语法糖
gather(A, B, C)本质上就是同时await A、await B、await C，只是一行搞定。
2.5 异常处理
# 单个协程：正常 try/except
async def main():
try:
await 不稳定的任务()
except ValueError as e:
print(f"捕获到：{e}")
# gather 中让各任务独立，互不影响
results = await asyncio.gather(
任务("A", 1),
不稳定的任务(),
任务("C", 2),
return_exceptions=True # 异常当返回值，不中断其他任务
)
for r in results:
if isinstance(r, Exception):
print(f"失败：{r}")
else:
print(f"成功：{r}")2.6 超时控制
async def main():
try:
result = await asyncio.wait_for(慢任务(), timeout=3) # 只等 3 秒
except asyncio.TimeoutError:
print("超时，放弃")2.7 取消任务
async def main():
task = asyncio.create_task(慢任务())
await asyncio.sleep(1)
task.cancel() # 取消
try:
await task
except asyncio.CancelledError:
print("任务已取消")三、常见错误
DANGER 错误一：忘记
awaitresult = 烧水() # 协程对象，代码没跑！ result = await 烧水() # ✅
DANGER 错误二：在普通函数里用
awaitdef 普通函数(): await 烧水() # SyntaxError！await 只能在 async def 里用
DANGER 错误三：用
time.sleep而不是asyncio.sleepasync def 烧水(): time.sleep(3) # 阻塞整个事件循环，其他协程全卡死！ await asyncio.sleep(3) # ✅
四、最小可运行模板
import asyncio
async def 任务(名字, 秒数):
print(f"{名字} 开始")
await asyncio.sleep(秒数)
print(f"{名字} 完成")
return f"{名字}的结果"
async def main():
results = await asyncio.gather(
任务("A", 3),
任务("B", 1),
任务("C", 2),
)
print(results)
asyncio.run(main())五、知识地图
asyncio.run(main()) ← 唯一入口，启动事件循环
└── async def main():
├── await 协程() 串行，等完再继续
├── asyncio.gather(...) 并发，等全部完成
│ └── return_exceptions=True 各自独立
├── asyncio.create_task(...) 并发，可单独控制
│ └── task.cancel() 取消
└── asyncio.wait_for(..., timeout=N) 加超时六、相关标签
#Python #异步编程 #asyncio #协程 #并发