第 3 章：并发任务组织：gather() 与 TaskGroup
我们继续做“学生信息查询助手”。
第 2 章我们已经能写出一个清晰的异步入口：
async def main():
await print_student(1)
if __name__ == "__main__":
asyncio.run(main())这个写法可以查询一个学生。
现在需求变了：用户一次想查询多个学生的信息。
按照上一章的写法，我们很容易写成这样：
async def main():
await print_student(1)
await print_student(2)
await print_student(3)这段代码能运行，也能查到结果。
但它有个明显问题：第 2 个查询要等第 1 个查询结束，第 3 个查询要等第 2 个查询结束。
如果每个查询都在等网络、数据库或外部服务，这样就没有发挥异步的优势。
本章要解决 4 个问题：
- 多个
await连着写，为什么还是顺序等待？ - 如何用
asyncio.gather()同时等待多个异步查询？ gather()返回的结果顺序应该怎么看？- 如何用
TaskGroup把多个相关任务组织得更清晰？
本章只解决“多个异步任务如何同时等待”。
本章不解决“最多同时跑多少个任务”。这个问题第 4 章再讲。
本章也不系统讲超时、取消和复杂异常处理。这些放到第 5 章。
本章学习目标
学完本章后，你应该能做到：
- 理解连续写多个
await时，代码仍然会按顺序等待。 - 知道
asyncio.gather()可以把多个异步操作放在一起等待。 - 能使用
asyncio.gather()同时查询多个学生信息。 - 能理解
gather()返回结果的顺序和传入顺序一致。 - 能使用
TaskGroup组织多个相关异步任务。 - 能初步判断什么时候用
gather()，什么时候用TaskGroup。 - 能排查忘记
await asyncio.gather(...)、gather()参数写错、提前读取task.result()等常见错误。
多学生查询场景
我们的学生信息查询助手目前只有一个内存字典：
student_data = {
1: {"id": 1, "name": "小明", "age": 13},
2: {"id": 2, "name": "小红", "age": 14},
3: {"id": 3, "name": "小刚", "age": 13},
}每次查询学生信息时，我们用 asyncio.sleep(1) 模拟外部服务等待 1 秒。
注意：这里不是真的查数据库，也不是真的请求 HTTP 接口。我们只是用 asyncio.sleep(1) 模拟“程序正在等外部结果”。
现在要查 3 个学生。
顺序等待与同时等待的流程对比如下：
本章的重点不是“异步一定更快”。
重点是：当多个任务都在等 I/O 结果时，我们可以让这些等待重叠发生。
并发查询方式
下面每一步的完整代码都直接显示在页面里。将代码按标注文件名保存到本地 _code_async/c03_gather_and_taskgroup/，然后在 _code_async 目录执行命令。
1. 连续 await 与顺序等待
await 的意思是：等这个异步操作完成，再继续往下执行。
所以如果代码这样写：
await print_student(1)
await print_student(2)
await print_student(3)它的执行顺序就是：
- 等学生 1 查询完成。
- 再等学生 2 查询完成。
- 再等学生 3 查询完成。
这段代码没有错。
它只是顺序等待。
日常项目中，如果后一个查询必须依赖前一个查询的结果，就可以这样写。
但如果 3 个查询互不依赖，就不适合一个一个等。
需求：一次查询 3 个学生，但先按照第 2 章的写法，一个一个 await。
功能分析：
- 准备一个内存字典
student_data。 - 用
query_student_async()模拟查询一个学生。 - 用
print_student()打印查询结果。 - 在
main()中连续等待 3 次。 - 用
time.perf_counter()观察总耗时。
文件名：01_sequential_await.py
import asyncio
import time
student_data = {
1: {"id": 1, "name": "小明", "age": 13},
2: {"id": 2, "name": "小红", "age": 14},
3: {"id": 3, "name": "小刚", "age": 13},
}
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 print_student(student_id):
result = await query_student_async(student_id)
if result is None:
print(f"没有找到 ID 为 {student_id} 的学生")
return
print(result)
async def main():
start_time = time.perf_counter()
await print_student(1)
await print_student(2)
await print_student(3)
end_time = time.perf_counter()
print(f"总耗时：{end_time - start_time:.2f} 秒")
if __name__ == "__main__":
asyncio.run(main())运行命令：
uv run python c03_gather_and_taskgroup/01_sequential_await.py可能看到类似结果：
开始查询学生 1
查询学生 1 完成
{'id': 1, 'name': '小明', 'age': 13}
开始查询学生 2
查询学生 2 完成
{'id': 2, 'name': '小红', 'age': 14}
开始查询学生 3
查询学生 3 完成
{'id': 3, 'name': '小刚', 'age': 13}
总耗时：3.01 秒运行逻辑：
main()先执行await print_student(1)。- 学生 1 查询完成后，才会执行
await print_student(2)。 - 学生 2 查询完成后，才会执行
await print_student(3)。 - 每次查询都等待 1 秒，所以总耗时接近 3 秒。
注意：这段代码是正确的。
问题不是“不能运行”，而是“互不依赖的等待没有重叠”。
如果 3 个查询之间没有依赖关系，我们可以让它们同时开始等待。
2. gather() 与并发结果收集
asyncio.gather() 可以先简单理解为：
把多个异步操作放在一起等待，等它们都完成后，一次性拿到结果。
常见写法是：
results = await asyncio.gather(
query_student_async(1),
query_student_async(2),
query_student_async(3),
)注意两点：
- 外面要写
await。 results的顺序和传入顺序一致。
也就是说，第一个结果对应 query_student_async(1)，第二个结果对应 query_student_async(2)，第三个结果对应 query_student_async(3)。
它不按谁先完成来排列。
要验证这一点，可以在 02_gather_query.py 中把查询函数的等待改为 await asyncio.sleep({1: 0.3, 2: 0.1, 3: 0.2}[student_id])。三个查询仍按 1、2、3 的顺序传入，但完成日志会依次显示学生 2、3、1；返回列表仍依次对应学生 1、2、3。总耗时接近最长的一次等待，即 0.3 秒，而不是三次等待之和。
先记住一个会影响项目写法的区别：默认情况下，一个操作抛出异常，gather() 会把异常立即交给调用方，但其他操作不会因此自动取消，它们仍可能继续运行。第 5 章再练习具体的异常处理。
协程对象列表与参数解包
如果学生 ID 放在列表里，常见写法是先生成一组协程对象：
student_ids = [1, 2, 3]
tasks = [query_student_async(student_id) for student_id in student_ids]这里变量名虽然叫 tasks，列表元素此时仍是协程对象，还不是已经运行的 Task。gather() 接收它们后，才会安排这些查询一起执行。
然后传给 gather()：
results = await asyncio.gather(*tasks)这里的 *tasks 是 Python 的解包语法。
它的作用是把列表里的多个元素展开，分别传给 gather()。
不要写成：
results = await asyncio.gather(tasks)这样传进去的是一个列表本身，不是多个异步操作。
需求：一次查询 3 个学生，并让 3 个查询同时开始等待。
功能分析：
- 保留
query_student_async()。 - 在
main()中创建 3 个异步查询。 - 用
await asyncio.gather(...)一起等待它们完成。 - 打印结果列表。
- 观察耗时接近 1 秒。
文件名：02_gather_query.py
import asyncio
import time
student_data = {
1: {"id": 1, "name": "小明", "age": 13},
2: {"id": 2, "name": "小红", "age": 14},
3: {"id": 3, "name": "小刚", "age": 13},
}
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 main():
start_time = time.perf_counter()
results = await asyncio.gather(
query_student_async(1),
query_student_async(2),
query_student_async(3),
)
end_time = time.perf_counter()
print("查询结果：")
print(results)
print(f"总耗时：{end_time - start_time:.2f} 秒")
if __name__ == "__main__":
asyncio.run(main())运行命令：
uv run python c03_gather_and_taskgroup/02_gather_query.py可能看到类似结果：
开始查询学生 1
开始查询学生 2
开始查询学生 3
查询学生 1 完成
查询学生 2 完成
查询学生 3 完成
查询结果：
[{'id': 1, 'name': '小明', 'age': 13}, {'id': 2, 'name': '小红', 'age': 14}, {'id': 3, 'name': '小刚', 'age': 13}]
总耗时：1.00 秒运行逻辑：
query_student_async(1)、query_student_async(2)、query_student_async(3)都是异步查询。asyncio.gather()把它们放在一起等待。- 3 个查询都在等待
asyncio.sleep(1)时，等待时间可以重叠。 - 所以总耗时接近 1 秒，而不是 3 秒。
results的顺序和传入顺序一致。
注意：
results = await asyncio.gather(
query_student_async(1),
query_student_async(2),
query_student_async(3),
)这里最容易漏掉的是外面的 await。
如果学生 ID 已经放在列表里，可以这样写：
student_ids = [1, 2, 3]
tasks = [query_student_async(student_id) for student_id in student_ids]
results = await asyncio.gather(*tasks)*tasks 表示把列表里的多个协程对象展开传入 gather()。
本章先掌握这个写法即可。异常处理后面再讲。
3. TaskGroup 与任务生命周期
前一个示例需要收集全部查询结果，gather() 已经够用。如果需求进一步变成“某个查询失败后，取消并等待尚未完成的同组查询”，就需要考虑这批任务的整体生命周期。
例如，学生信息聚合必须同时取得基本信息、成绩和课程表；其中一个查询失败后，剩余查询即使完成，也无法交付完整结果。TaskGroup 可以统一管理这组任务的失败。本章先理解这一选型依据，异常捕获与取消的具体处理留到第 5 章。
TaskGroup 可以先简单理解为：
用一个任务组管理一批相关异步任务。
够用写法是：
tasks = []
async with asyncio.TaskGroup() as tg:
for student_id in student_ids:
task = tg.create_task(query_student_async(student_id))
tasks.append(task)
results = [task.result() for task in tasks]注意：
tg.create_task()会把协程加入任务组。async with正常退出时，任务组里的任务都已经成功执行完成。- 所以
task.result()要放在async with代码块结束后读取。
简单收集结果，可以先用 gather()。如果必须等一组操作全部结束后再处理失败，可以使用 gather(..., return_exceptions=True)，然后逐个检查结果；异常此时也会占据一个结果位置。
TaskGroup 适合一组任务需要统一管理失败的场景。组中某个任务抛出普通异常时，它会取消尚未完成的同组任务，等待它们结束，再向外抛出 ExceptionGroup。已经完成的外部操作不会自动回滚。本章只要求理解这一行为，不展开异常组的高级用法。
本章示例没有主动取消任务。若任务失败导致任务组向外抛出异常，程序不会直接执行后面的结果收集语句；被取消的任务也不能当作成功结果读取。
需求：还是查询 3 个学生，但这次用 TaskGroup 把这 3 个查询放到同一个任务组里。
功能分析：
- 准备
student_ids = [1, 2, 3]。 - 使用
async with asyncio.TaskGroup() as tg创建任务组。 - 在任务组中使用
tg.create_task()添加查询任务。 - 保存每个任务对象。
- 任务组结束后，再读取每个任务的结果。
文件名：03_task_group.py
import asyncio
import time
student_data = {
1: {"id": 1, "name": "小明", "age": 13},
2: {"id": 2, "name": "小红", "age": 14},
3: {"id": 3, "name": "小刚", "age": 13},
}
student_ids = [1, 2, 3]
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 main():
start_time = time.perf_counter()
tasks = []
async with asyncio.TaskGroup() as tg:
for student_id in student_ids:
task = tg.create_task(
query_student_async(student_id),
name=f"query-student-{student_id}",
)
tasks.append(task)
results = [task.result() for task in tasks]
end_time = time.perf_counter()
print("查询结果：")
print(results)
print(f"总耗时：{end_time - start_time:.2f} 秒")
if __name__ == "__main__":
asyncio.run(main())运行命令：
uv run python c03_gather_and_taskgroup/03_task_group.py可能看到类似结果：
开始查询学生 1
开始查询学生 2
开始查询学生 3
查询学生 1 完成
查询学生 2 完成
查询学生 3 完成
查询结果：
[{'id': 1, 'name': '小明', 'age': 13}, {'id': 2, 'name': '小红', 'age': 14}, {'id': 3, 'name': '小刚', 'age': 13}]
总耗时：1.00 秒运行逻辑：
async with asyncio.TaskGroup() as tg创建一个任务组。tg.create_task()把每个学生查询加入任务组。- 这些任务会在任务组中一起运行。
async with正常退出时，本例中的任务都已成功完成。- 所以我们在代码块外面用
task.result()读取结果。
注意：不要在 async with 代码块内部提前读取 task.result()。
因为任务可能还没有完成。
name=f"query-student-{student_id}" 是给任务起名字。这里知道可以命名即可，后面做排查时会更有用。
并发工具选择
日常项目里可以先这样判断：
| 场景 | 推荐写法 |
|---|---|
| 简单并发查询，并按传入顺序收集结果 | asyncio.gather() |
| 一组任务需要统一管理失败，并取消尚未完成的同组任务 | asyncio.TaskGroup |
| 一组任务需要全部结束后再统一检查错误 | asyncio.gather(..., return_exceptions=True) |
| 需要严格控制并发数量 | 第 4 章学习 Semaphore |
| 需要处理超时、取消、异常细节 | 第 5 章专门处理 |
本章只先把“同时等待多个任务”的基本写法掌握住。
4. 批量学生信息查询
本章完整案例做一个普通 Python 脚本。
它能批量查询多个学生，并用 asyncio.gather() 同时等待多个查询结果。
项目结构
_code_async/c03_gather_and_taskgroup/
└── 04_batch_query.py完整代码
文件名：04_batch_query.py
import asyncio
import time
student_data = {
1: {"id": 1, "name": "小明", "age": 13},
2: {"id": 2, "name": "小红", "age": 14},
3: {"id": 3, "name": "小刚", "age": 13},
4: {"id": 4, "name": "小丽", "age": 15},
}
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)
def format_student(student):
if student is None:
return "没有找到学生"
return f"ID：{student['id']}，姓名：{student['name']}，年龄：{student['age']}"
async def query_students_with_gather(student_ids):
tasks = [query_student_async(student_id) for student_id in student_ids]
results = await asyncio.gather(*tasks)
return results
async def query_students_with_task_group(student_ids):
tasks = []
async with asyncio.TaskGroup() as tg:
for student_id in student_ids:
task = tg.create_task(
query_student_async(student_id),
name=f"query-student-{student_id}",
)
tasks.append(task)
return [task.result() for task in tasks]
async def main():
student_ids = [1, 2, 3, 99]
start_time = time.perf_counter()
results = await query_students_with_gather(student_ids)
end_time = time.perf_counter()
print("\n批量查询结果：")
for student_id, student in zip(student_ids, results):
print(f"学生 ID {student_id}：{format_student(student)}")
print(f"\n总耗时：{end_time - start_time:.2f} 秒")
if __name__ == "__main__":
asyncio.run(main())运行命令
从本地 _code_async 目录运行脚本：
uv run python c03_gather_and_taskgroup/04_batch_query.py运行结果
可能看到类似结果：
开始查询学生 1
开始查询学生 2
开始查询学生 3
开始查询学生 99
查询学生 1 完成
查询学生 2 完成
查询学生 3 完成
查询学生 99 完成
批量查询结果：
学生 ID 1：ID：1，姓名：小明，年龄：13
学生 ID 2：ID：2，姓名：小红，年龄：14
学生 ID 3：ID：3，姓名：小刚，年龄：13
学生 ID 99：没有找到学生
总耗时：1.00 秒说明：
- 4 个查询都同时开始等待。
- 每个查询等待 1 秒。
- 总耗时接近 1 秒，而不是 4 秒。
student_id = 99查不到，所以返回None，再由format_student()转成友好提示。query_students_with_task_group()是本章对照写法，完整案例主流程先使用gather()。
注意：实际耗时可能是 1.00、1.01 或略有浮动，这是正常现象。
并发调用错误排查
| 现象 | 原因 | 修复方式 |
|---|---|---|
打印出 gather() 返回的 Future，而不是学生列表；脚本退出时还可能出现取消相关提示 | 遗漏了 await asyncio.gather(...) 中的 await。gather() 已安排查询执行，但当前流程没有等待其结果；这与仅创建一个未运行的协程对象不同。 | 用 results = await asyncio.gather(...) 等待并取得结果。 |
TypeError: cannot use 'list' as a dict key (unhashable type: 'list') | 写成 asyncio.gather(tasks)，把普通列表作为一个参数传入；创建的查询协程没有被接管，还可能出现未等待警告。 | 使用 asyncio.gather(*tasks) 解包协程对象列表。 |
InvalidStateError: Result is not set.（在本例中包含于 ExceptionGroup） | 在任务完成前调用 task.result()。 | 在任务组正常退出后读取本例的任务结果。 |
本章小结
本章我们解决了“多个异步查询如何同时等待”的问题。
核心结论有 4 个：
- 连续写多个
await，代码仍然会按顺序等待。 asyncio.gather()可以同时等待多个异步操作，并按传入顺序返回结果。- 如果异步操作来自列表，要用
*tasks解包传给gather()。 TaskGroup会在一个任务失败时取消并等待同组其他任务；读取task.result()要放在任务组结束后。
本章先掌握“同时等待”。
任务数量过多时怎么控制，第 4 章继续讲。
练习
练习 1：先把 3 个查询写成连续 await，记录一次耗时；再改成 asyncio.gather()，对比两次耗时。
练习 2：在 02_gather_query.py 中，按正文把学生 1、2、3 的等待时间分别改为 0.3、0.1、0.2 秒。检查完成日志是否按 2、3、1 排列，而结果列表仍按 1、2、3 排列。再在 04_batch_query.py 中把 student_ids 改为 [99, 1, 2, 3]，确认“没有找到学生”位于结果第一项。
练习 3：把本章的 gather() 版本再手动改写成 TaskGroup 版本，体会“收集结果”和“组织任务”这两个关注点的区别。
练习 4：分别验证错误排查表中的三种写法，每次只改一处，观察后恢复：① 在 02_gather_query.py 中去掉 gather() 前的 await，观察 results 是否变成 Future；② 在 04_batch_query.py 中把 gather(*tasks) 改成 gather(tasks)，观察列表参数错误；③ 在 03_task_group.py 的 tasks.append(task) 后立即加入 print(task.result())，观察结果未就绪错误。修复后确认各示例重新打印学生结果。