第 1 章：异步编程与 I/O 等待

我们正在做一个“学生信息查询助手”。

现在要查询 3 个学生的信息。三次查询互不依赖，每次都要等 1 秒，模拟访问外部接口、数据库或 Redis 的耗时。

如果一个一个查，总共约需 3 秒。能否让三段等待重叠，把总耗时降到接近 1 秒？

我们会先查询一个学生，再连续查询三个学生，最后改成异步查询。运行每个版本，回答三个问题：

1. 为什么同步查询多个 I/O 等待任务会变慢。
2. 异步代码和同步代码在等待场景中有什么区别。
3. async def、await、asyncio.run() 在最小异步程序里怎么用。

本章只比较等待方式，并跑通最小异步示例。并发任务的组织、异常、超时、并发数量限制、真实 HTTP 请求和 Web 接口开发留到后续章节。

本章学习目标

学完本章，你要能做到：

1. 用 time.perf_counter() 测出连续查询时等待时间怎样累加。
2. 读懂 async def、await asyncio.sleep() 和 asyncio.run() 的最小写法。
3. 对比三个查询的同步与异步运行结果，说明等待重叠需要什么条件。
4. 判断 I/O 等待与 CPU 计算场景中，异步能解决什么问题。

学生信息查询场景

我们的统一案例叫“异步学生信息聚合与任务处理系统”。

当前系统还只是 Python 脚本，没有 Web 服务或真实数据库。每次查询都人为等待 1 秒，代表一次外部查询的等待时间。

这 1 秒可以理解为：

1. 等数据库返回结果。
2. 等 Redis 返回结果。
3. 等外部 HTTP 接口返回结果。
4. 等文件系统或网络 I/O 返回结果。

注意，这 1 秒里程序大部分时间不是在计算，而是在等。

同步查询的流程大致是：

这个流程的问题很直接：每一次等待都会排队。

异步查询希望达到的效果是：

异步版本要让三个查询都开始，并在各自等待时给其他查询执行机会。下面用可运行的代码验证两张图中的耗时差异。

同步与异步查询对比

下面每一步的完整代码都直接显示在页面里。将代码按标注文件名保存到本地 _code_async/c01_why_async_programming/，然后在 _code_async 目录执行命令。

1. 单次同步查询

先查询一个学生。这里的 time.sleep(1) 会让当前线程暂停 1 秒，用来模拟一次同步查询的等待。

需求：查询一个学生的信息，等待 1 秒后返回结果。

功能分析：

1. 准备一份内存学生数据。
2. 写一个普通函数 query_student_sync(student_id)。
3. 在函数里用 time.sleep(1) 模拟等待。
4. 根据学生 ID 返回学生信息。

文件名：01_sync_query.py

01_sync_query.py

import time

student_data = {
    1: {"id": 1, "name": "张三", "age": 18},
    2: {"id": 2, "name": "李四", "age": 19},
    3: {"id": 3, "name": "王五", "age": 20},
}


def query_student_sync(student_id):
    print(f"开始查询学生：{student_id}")
    time.sleep(1)
    print(f"查询完成：{student_id}")
    return student_data.get(student_id)


result = query_student_sync(1)
print(result)

运行命令：

uv run python c01_why_async_programming/01_sync_query.py

可能看到的输出：

开始查询学生：1
查询完成：1
{'id': 1, 'name': '张三', 'age': 18}

运行逻辑：

1. 程序先执行 query_student_sync(1)。
2. 函数内部打印“开始查询学生：1”。
3. time.sleep(1) 让执行脚本的当前线程暂停 1 秒。
4. 1 秒后，程序继续往下执行。
5. 函数返回学生数据。
6. 最后打印查询结果。

注意：数据来自内存字典，time.sleep(1) 只是在模拟“等待外部结果”，并未执行真实 I/O。真实项目中，数据库、Redis、网络请求或文件读写都可能产生 I/O 等待；程序主要在等结果，而不是进行 CPU 计算。看到程序慢时，先分清它是在计算还是在等待。

在这个脚本里，当前查询没结束，后面的查询代码就不会开始。time.sleep() 也适用于需要暂停的简单同步脚本；放进后面的异步查询函数，则会让其他查询无法趁这 1 秒推进。练习 3 会验证这一点。

2. 连续同步查询与耗时累加

现在把需求改成一次查询 3 个学生的信息。同步版本按顺序调用查询函数：前一个学生查完，才开始查下一个。

功能分析：

1. 准备学生 ID 列表。
2. 用 for 循环一个一个查询。
3. 每个学生查询等待 1 秒。
4. 用 time.perf_counter() 统计总耗时。

文件名：02_sync_query.py

02_sync_query.py

import time

student_data = {
    1: {"id": 1, "name": "张三", "age": 18},
    2: {"id": 2, "name": "李四", "age": 19},
    3: {"id": 3, "name": "王五", "age": 20},
}


def query_student_sync(student_id):
    print(f"开始查询学生：{student_id}")
    time.sleep(1)
    print(f"查询完成：{student_id}")
    return student_data.get(student_id)


def query_students_sync(student_ids):
    results = []

    for student_id in student_ids:
        student = query_student_sync(student_id)
        results.append(student)

    return results


student_ids = [1, 2, 3]

start_time = time.perf_counter()
results = query_students_sync(student_ids)
end_time = time.perf_counter()

print(results)
print(f"同步查询耗时：{end_time - start_time:.2f} 秒")

运行命令：

uv run python c01_why_async_programming/02_sync_query.py

可能看到的输出：

开始查询学生：1
查询完成：1
开始查询学生：2
查询完成：2
开始查询学生：3
查询完成：3
[{'id': 1, 'name': '张三', 'age': 18}, {'id': 2, 'name': '李四', 'age': 19}, {'id': 3, 'name': '王五', 'age': 20}]
同步查询耗时：3.01 秒

运行逻辑：

1. student_ids = [1, 2, 3] 表示要查询 3 个学生。
2. start_time = time.perf_counter() 记录开始时间。
3. query_students_sync(student_ids) 开始同步查询多个学生。
4. for 循环先查学生 1，等待 1 秒。
5. 学生 1 查询完成后，才开始查学生 2。
6. 学生 2 查询完成后，才开始查学生 3。
7. 3 个学生全部查完后，记录结束时间。
8. 最后打印总耗时。

为什么接近 3 秒？

因为每次查询都要等 1 秒，并且 3 次查询是顺序执行的。

学生 1 等 1 秒 + 学生 2 等 1 秒 + 学生 3 等 1 秒 ≈ 3 秒

注意：这里的时间主要花在等待上，不是 Python 在计算学生字典。任务很少、等待很短或同步代码已经足够清楚时，按顺序查询仍是合适的写法；当大量慢查询都这样排队，总耗时就会继续累加。

3. 异步查询与等待重叠

同步版本已经测出约 3 秒。现在希望三个查询都开始等待，让三段等待尽量重叠。异步查询函数用 async def 定义；执行到 await asyncio.sleep(1) 时，它会暂停当前查询，让其他查询有机会推进。这里的 sleep() 只模拟异步等待，在实际异步程序中也可用于重试或轮询间隔。

还需要让多个查询一起开始：本章借 asyncio.gather() 完成这一步，并等它们全部返回。asyncio.run(main()) 从普通脚本启动异步主流程。先把这两个写法当作运行示例所需的工具；异步函数调用后的返回对象与 await 规则留到第 2 章，多个任务如何组织留到第 3 章。

功能分析：

1. 使用 async def 定义异步查询函数。
2. 使用 await asyncio.sleep(1) 模拟异步等待。
3. 使用 asyncio.gather() 同时等待多个异步查询结果。
4. 使用 asyncio.run(main()) 运行异步入口函数。
5. 使用 time.perf_counter() 统计总耗时。

文件名：03_async_query.py

03_async_query.py

import asyncio
import time

student_data = {
    1: {"id": 1, "name": "张三", "age": 18},
    2: {"id": 2, "name": "李四", "age": 19},
    3: {"id": 3, "name": "王五", "age": 20},
}


async def query_student_async(student_id):
    print(f"开始查询学生：{student_id}")
    await asyncio.sleep(1)
    print(f"查询完成：{student_id}")
    return student_data.get(student_id)


async def query_students_async(student_ids):
    results = await asyncio.gather(
        query_student_async(student_ids[0]),
        query_student_async(student_ids[1]),
        query_student_async(student_ids[2]),
    )
    return results


async def main():
    student_ids = [1, 2, 3]

    start_time = time.perf_counter()
    results = await query_students_async(student_ids)
    end_time = time.perf_counter()

    print(results)
    print(f"异步查询耗时：{end_time - start_time:.2f} 秒")


asyncio.run(main())

运行命令：

uv run python c01_why_async_programming/03_async_query.py

可能看到的输出：

开始查询学生：1
开始查询学生：2
开始查询学生：3
查询完成：1
查询完成：2
查询完成：3
[{'id': 1, 'name': '张三', 'age': 18}, {'id': 2, 'name': '李四', 'age': 19}, {'id': 3, 'name': '王五', 'age': 20}]
异步查询耗时：1.01 秒

运行逻辑：

1. async def query_student_async(student_id) 定义一个异步查询函数。
2. asyncio.gather() 让三个查询都有机会开始执行，并等待全部结果。
3. 每个查询执行到 await asyncio.sleep(1) 时，暂停自己，让其他查询推进。
4. async def main() 组织查询和计时；asyncio.run(main()) 从脚本入口启动它。
5. 查询全部完成后打印结果和总耗时。

为什么 3 个学生大约只要 1 秒？

因为 gather() 让三个查询都开始了；每个查询在异步等待时让出执行机会，三段等待便能够重叠。

学生 1 等 1 秒
学生 2 等 1 秒
学生 3 等 1 秒
三段等待重叠，所以总耗时接近 1 秒

只写 async def 不会自动让多个查询同时推进。如果在循环里逐个 await query_student_async(student_id)，仍是完成一个再等下一个，总耗时接近 3 秒；本例用 gather() 让等待重叠。第 3 章再专门讲多个任务的组织、结果收集和异常影响。

等待能重叠，还因为这三个查询互不依赖。如果查询学生 2 必须先拿到学生 1 的返回值，第二次查询就只能等第一次完成后再开始；即使用了异步，有依赖的两段等待也不能完全重叠。

这也是模拟实验：它说明了三段人为等待可以重叠，并不保证真实接口一定能从 3 秒降到 1 秒。真实耗时还取决于网络、服务端响应和并发限制。

常见误用是把 await asyncio.sleep(1) 改成 time.sleep(1)，或忘记在异步调用前写 await。前者会阻塞运行查询的事件循环；后者涉及异步函数的调用规则，第 2 章会通过运行结果专门讲解。普通 .py 脚本也不能直接在顶层写 await，所以本例把异步流程放进 main()，由 asyncio.run(main()) 启动。

异步的适用边界与代码写法

适合异步的 I/O 等待场景

异步适合用在 I/O 等待比较多的场景：

1. 一个接口要请求多个外部服务。
2. 一个任务要等待多个远程接口返回。
3. Web 服务里有大量网络 I/O。
4. 后台任务经常访问数据库、Redis、HTTP 接口。
5. 爬虫需要同时等待多个网页响应。

这些场景的共同点是：程序经常在等外部结果。

异步收益有限的场景

异步不是所有“慢问题”的通用答案。

下面这些场景不要一上来就用 async：

1. 主要耗时是大量数学计算。
2. 主要耗时是图像处理、视频编码、模型推理。
3. 代码只有很少的 I/O 等待。
4. 普通同步代码已经足够简单稳定。

如果慢是 CPU 算出来的，async 本身不会直接让 CPU 算得更快。

采用异步还需要正确处理资源释放、超时和异常。团队尚未掌握这些写法时，也要把学习和维护成本算进去。

最小异步程序的写法

本章阶段先记住几条：

1. 异步函数用 async def 定义。
2. 异步等待用 await。
3. 异步程序入口用 asyncio.run(main())。
4. 异步代码中优先使用异步等待，例如 await asyncio.sleep()。
5. 做耗时对比时，用 time.perf_counter()。

常见误用

先避免这些常见写法：

1. 在 async def 里写 time.sleep()。
2. 调用异步函数后忘记 await。
3. 写了 async def main()，但没有用 asyncio.run(main()) 启动。
4. 看到 async 就以为代码一定更快。
5. 本来只有一个简单查询，也强行改成异步。

异步是工具，不是项目变快的保证。

本章小结

本章我们用“查询多个学生信息”这个小需求，直观看到了同步和异步的差异。

同步代码会一个一个等，所以等待时间会累加。

当多个异步查询一起开始，并在等待时让出执行机会，它们的 I/O 等待就可以重叠。

本章最核心的是 4 个写法：

1. async def：定义异步函数。
2. await asyncio.sleep(1)：模拟异步等待，并让其他查询有机会推进。
3. asyncio.gather()：在本章示例中让三个查询一起开始，具体用法留到第 3 章。
4. asyncio.run(main())：运行异步程序入口。

异步适合处理 I/O 等待，不是用来直接加速所有代码。

练习

练习 1：把同步版本里的学生 ID 改成 [1, 2, 3, 1, 2]，自己运行一次，看看总耗时会不会继续累加。

练习 2：把异步版本里的等待时间改成 2 秒，再运行一次，观察总耗时是不是接近 2 秒，而不是 6 秒。

练习 3：尝试把异步函数里的 await asyncio.sleep(1) 故意改成 time.sleep(1)，再观察输出顺序和总耗时。你会更容易理解“阻塞”是什么意思。

练习 4：判断下面三个任务中，哪些等待可以重叠、哪些步骤必须按顺序执行、哪个任务不会因为改成 async 就直接变快，并说明原因：① 分别从外部接口查询 3 个学生的基本信息，查询之间互不依赖；② 先查询学生的班级 ID，再用这个 ID 查询班级课程表；③ 对内存中的大量分数求平均值。