📖 Day 1 学习页:异步编程核心
Day 1:异步编程核心
🎯 今日目标:今天结束时,你能写出"并发调用 5 个 API、总耗时只等于最慢那一个"的异步代码, 并理解为什么它是 Agent 并发调用工具的基石。
⏱ 今日安排(约 2 小时)
| 时间 | 内容 | 方式 |
|---|---|---|
| 40min | 理论学习:本页「第一步」+ 跑一遍异步教程 | 阅读 + 运行 |
| 50min | 动手实操:完成练习模板中的 5 个任务 | 编码 |
| 30min | 验收巩固:跑验收脚本,按提示修到 80 分以上 | 编码 |
📖 第一步:理论学习(先看懂,再动手)
1. 同步 vs 异步:烧水做饭的比喻
同步就是"一件做完才做下一件";异步是"发起后先不等,谁先完成先处理谁"。 烧水要 3 秒、切菜要 2 秒:同步总共 5 秒,异步只需要 3 秒(等水开的同时把菜切了):
import asyncio
import time
# 同步:一件一件来(总耗时 = 所有任务时间之和)
def 同步烧水():
time.sleep(3) # ❌ time.sleep 会阻塞整个程序,什么都干不了
return "开水"
# 异步:等待时把 CPU 让给其他任务(总耗时 ≈ 最慢的那个任务)
async def 异步烧水():
await asyncio.sleep(3) # ✅ await 时其他任务可以运行
return "开水"
async def 异步做饭():
任务1 = asyncio.create_task(异步烧水()) # 立刻开始烧水
任务2 = asyncio.create_task(异步切菜()) # 同时开始切菜
水, 菜 = await 任务1, await 任务2 # 都完成后取结果
return 水, 菜
# 结果:同步 5 秒,异步 3 秒!
2. 必须记住的 6 个 API
| 关键字 | 作用 | 示例 |
|---|---|---|
async def | 定义协程函数 | async def 下载文件(): |
await | 等待异步操作(不阻塞其他任务) | result = await asyncio.sleep(1) |
asyncio.run() | 启动事件循环(整个程序只调一次) | asyncio.run(main()) |
asyncio.gather() | 并发执行多个任务,等全部完成 | await asyncio.gather(t1, t2) |
asyncio.create_task() | 把协程包装成后台任务立刻调度 | task = asyncio.create_task(下载()) |
asyncio.timeout() | 超时控制(Python 3.11+) | async with asyncio.timeout(5): |
3. gather 和 create_task 的区别(面试常问)
gather(t1, t2):等着这批一起完成,按顺序返回结果列表 —— 适合"同时发 N 个请求然后汇总"。create_task(协程):扔到后台立刻跑,返回 task 对象,你可以先干别的再 await 它 —— 适合"边等下载边响应用户"。
4. Agent 视角:为什么 Agent 离不开异步
Agent 一次经常要并发调用多个工具(查天气 + 算数 + 搜文件)。逐个 await 是 4 倍耗时,gather 只要最长的那个:
async def Agent并发调用():
结果们 = await asyncio.gather(
调用工具("计算器", "3 + 5 * 2"),
调用工具("天气查询", "北京"),
调用工具("文件搜索", "config.py"),
调用工具("翻译", "Hello World"),
) # 4 个工具只花了最长的那个的时间(不是 4 倍!)
🛠 第二步:动手实操(核心环节)
- 跑教程:运行
python tutorial_02_async.py,跟着 9 步输出逐步理解(第 5 步故意演示 time.sleep 陷阱,别跳过)。 - 打开模板:day01_practice_template.py,共 5 个 TODO,函数体全部留空等你实现:
| 任务 | 要做什么 | 提示 |
|---|---|---|
| 任务 1 | 实现 异步请求(url) | await asyncio.sleep(延迟秒) 后返回含 url/status/data 的字典 |
| 任务 2 | 实现 并发请求(urls) | 用 asyncio.gather(*tasks),不要在循环里逐个 await |
| 任务 3 | 实现 同步请求 和 性能对比 | 同步用 time.sleep(1)(故意慢),对比后断言异步快一倍以上 |
| 任务 4 | 实现 异步请求带重试 | for 循环 + try/except + random.random() < 0.3 模拟失败 |
- 运行自测:
python day01_practice_template.py,确认 4 个任务都打印出结果。 - 验收:运行
python day01_practice_validator.py。验收脚本会真的计时你的并发实现——逐个 await 的伪并发会被抓出来,80 分以上才算通过。
⚠️ 避坑(今天就会遇到的)
- ❌ 在 async 函数里用
time.sleep()—— 整个事件循环被卡死,并发变串行;永远用await asyncio.sleep() - ❌
asyncio.run()在一个程序里调用多次或嵌套调用 —— 它只能由主线程调用一次 - ✅ 批量并发优先用
asyncio.gather(),而不是先 await 一个再 await 下一个 - 💡 验收提示里的
random.random()需要import random——模板已帮你加好