📖 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 倍!)

🛠 第二步:动手实操(核心环节)

  1. 跑教程:运行 python tutorial_02_async.py,跟着 9 步输出逐步理解(第 5 步故意演示 time.sleep 陷阱,别跳过)。
  2. 打开模板: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 模拟失败
  1. 运行自测:python day01_practice_template.py,确认 4 个任务都打印出结果。
  2. 验收:运行 python day01_practice_validator.py。验收脚本会真的计时你的并发实现——逐个 await 的伪并发会被抓出来,80 分以上才算通过。

⚠️ 避坑(今天就会遇到的)

  • ❌ 在 async 函数里用 time.sleep() —— 整个事件循环被卡死,并发变串行;永远用 await asyncio.sleep()
  • ❌ asyncio.run() 在一个程序里调用多次或嵌套调用 —— 它只能由主线程调用一次
  • ✅ 批量并发优先用 asyncio.gather(),而不是先 await 一个再 await 下一个
  • 💡 验收提示里的 random.random() 需要 import random——模板已帮你加好

🔗 延伸资源