第 4 章：Semaphore 与并发控制
我们继续做“学生信息查询助手”。
前面已经能用 asyncio.gather() 同时查询多个学生。现在需求变大了：不只是查 3 个学生，而是一次查一个班、一个年级，甚至全校学生。
如果一次要查询 1000 个学生，直接把 1000 个异步查询全部交给 gather()，代码可以写出来：
tasks = [query_student_async(student_id) for student_id in range(1, 1001)]
results = await asyncio.gather(*tasks)这段代码本身没有语法错误。
问题出在真实项目里：
- 如果查询背后是数据库，可能把数据库连接打满。
- 如果查询背后是 Redis，可能让 Redis 瞬间压力过高。
- 如果查询背后是外部接口，可能触发对方接口限制。
- 如果任务数量太多，也会让自己的程序占用更多内存和调度资源。
所以本章要解决 4 个问题：
- 为什么不能让大量查询同时访问外部资源？
- 什么是并发数量？为什么真实项目里要控制它？
- 如何用
asyncio.Semaphore()限制同一时间运行的查询数量？ - 如何把并发限制封装到批量查询函数里，让代码更适合项目使用？
本章只解决“并发数量控制”。
不解决超时、异常、取消、重试、HTTP 请求、数据库连接池和 FastAPI 接口。这些后面再讲。
本章学习目标
学完本章，我们要能做到：
- 理解“同时等待多个任务”和“无限并发”的区别。
- 知道数据库、Redis、外部接口、文件系统等资源都有承载上限。
- 能使用
asyncio.Semaphore(limit)创建并发数量限制器。 - 能使用
async with semaphore:限制某段异步代码的同时运行数量。 - 能编写带并发限制的批量查询函数。
- 能排查
Semaphore常见错误，例如限制器创建位置错误、忘记async with、限制位置没有包住真正的 I/O 操作。
批量查询的资源压力
上一章我们已经能同时查询多个学生：
results = await asyncio.gather(
query_student_async(1),
query_student_async(2),
query_student_async(3),
)查询 3 个学生没有问题。
但是如果学生 ID 变成这样：
student_ids = range(1, 1001)这时就要先想一个问题：
能同时查，不代表应该全部同时查。真实项目里，查询学生信息可能会访问数据库、Redis、外部 HTTP 接口。它们都有自己的承载上限。
我们希望程序这样运行：
图示重点说明：
- 任务可以很多。
- 但同一时间真正进入查询区的任务数量要受限制。
Semaphore就是用来控制这个“进入名额”的。
并发限制与批量查询
下面每一步的完整代码都直接显示在页面里。将代码按标注文件名保存到本地 _code_async/c04_semaphore_limit_concurrency/，然后在 _code_async 目录执行命令。
1. 无限制查询与资源压力
并发数量
并发数量，可以先理解为：
同一时间进入某段关键操作的任务数量。比如我们限制并发数量为 5，就表示同一时间最多只允许 5 个查询进入真正的查询逻辑。
它解决的问题是：
- 避免一次性请求太多外部资源。
- 避免把数据库、Redis、外部接口压垮。
- 让程序更稳定，而不是只追求一瞬间跑得快。
容易误用的地方是：
- 以为异步任务越多越好。
- 以为
gather()能开多少就开多少。 - 只看本地脚本跑得快，不考虑外部服务能不能承受。
无限并发
无限并发不是一个严格的技术术语。这里我们用它表示：
没有任何限制地一次性提交大量异步任务。比如：
tasks = [query_student_async(student_id) for student_id in range(1, 100001)]
results = await asyncio.gather(*tasks)这种写法在教学脚本里可能能运行。
但真实项目里，如果每个任务都访问外部接口，就可能造成大量请求同时打出去。
注意：这段代码不是“绝对不能写”。
问题是：当任务背后连接真实资源时，必须考虑资源承载能力。
需求：先看看不做限制时，大量任务会怎样开始执行。
功能分析：
- 准备 100 个学生 ID。
- 每个学生查询用
asyncio.sleep()模拟等待。 - 使用
asyncio.gather()一次性等待所有查询。 - 用
running_count和max_running_count观察同时进入查询逻辑的任务数量。
文件名：01_unlimited_tasks.py
import asyncio
import time
running_count = 0
max_running_count = 0
async def query_student_async(student_id):
global running_count, max_running_count
running_count += 1
max_running_count = max(max_running_count, running_count)
if student_id <= 10:
print(f"开始查询学生 {student_id}，当前正在查询：{running_count}")
await asyncio.sleep(0.2)
running_count -= 1
return {"id": student_id, "name": f"学生{student_id}"}
async def main():
student_ids = range(1, 101)
start_time = time.perf_counter()
tasks = [query_student_async(student_id) for student_id in student_ids]
results = await asyncio.gather(*tasks)
end_time = time.perf_counter()
print(f"查询数量：{len(results)}")
print(f"观察到的最大同时查询数量：{max_running_count}")
print(f"总耗时：{end_time - start_time:.2f} 秒")
if __name__ == "__main__":
asyncio.run(main())运行命令：
uv run python c04_semaphore_limit_concurrency/01_unlimited_tasks.py可能看到类似结果：
开始查询学生 1，当前正在查询：1
开始查询学生 2，当前正在查询：2
开始查询学生 3，当前正在查询：3
...
开始查询学生 10，当前正在查询：10
查询数量：100
观察到的最大同时查询数量：100
总耗时：0.20 秒运行逻辑说明：
tasks = [...]会创建 100 个协程对象。await asyncio.gather(*tasks)会把这些查询一起提交并等待完成。- 每个查询进入函数后，都会先让
running_count加 1。 - 因为没有限制，很多查询会很快都进入等待状态。
- 示例里用的是
asyncio.sleep(0.2)，所以看起来很轻松。 - 如果这里换成真实数据库或外部接口，就可能造成压力。
注意：这段代码不是错误代码。
它的作用是引出一个项目问题：当任务数量变大时，我们需要控制同一时间进入外部查询区的数量。Semaphore 不会减少已创建的协程对象或排队任务；批量规模极大时，创建与排队本身仍会占用内存，需要另行考虑分批提交。
2. Semaphore 与受限查询区
名额申请与释放
asyncio.Semaphore(limit) 可以理解为一个异步名额控制器。
比如：
semaphore = asyncio.Semaphore(5)表示同一时间最多允许 5 个任务进入受限区域。其他任务可以已经创建，但会等名额。
这里限制的是同时占用名额的查询数量，不是每秒发起的请求次数。例如，每次查询只需要 0.1 秒，5 个名额在一秒内可能被反复使用，因此 Semaphore(5) 不代表“每秒最多请求 5 次”。外部接口若限制每秒请求次数，还需要另行控制请求频率；本章不展开频率限制的实现。
常见写法是：
async with semaphore:
await query_external_service()这段代码的意思是：
- 进入
async with前，先申请一个名额。 - 如果还有名额，就进入执行。
- 如果没有名额，就先等着。
- 离开
async with后，自动释放名额。
注意：Semaphore 只解决并发数量问题。
它不解决：
- 某个任务一直不返回。
- 查询失败。
- 网络超时。
- 任务取消。
- 接口重试。
这些是第 5 章要解决的问题。
受限区域的位置
async with semaphore: 要包住真正消耗外部资源的那段代码。
例如：
async with semaphore:
await asyncio.sleep(1)本章用 asyncio.sleep(1) 模拟外部 I/O 等待。
以后换成真实项目时，这里可能是：
- 查询数据库。
- 请求外部接口。
- 访问 Redis。
- 读取远程文件。
不要把 async with semaphore: 放在无关位置。
如果限制区域没有包住真正的 I/O 操作，限制就没有意义。
需求：让同一时间最多只有 5 个学生查询进入受限区域。
功能分析：
- 创建
asyncio.Semaphore(5)。 - 把真正需要限制的查询等待放进
async with semaphore:。 - 继续用
running_count观察同时运行数量。 - 使用
gather()等待所有任务完成。
文件名：02_semaphore_basic.py
import asyncio
import time
running_count = 0
max_running_count = 0
async def query_student_async(student_id, semaphore):
global running_count, max_running_count
async with semaphore:
running_count += 1
max_running_count = max(max_running_count, running_count)
print(f"开始查询学生 {student_id}，当前正在查询：{running_count}")
await asyncio.sleep(0.5)
print(f"结束查询学生 {student_id}")
running_count -= 1
return {"id": student_id, "name": f"学生{student_id}"}
async def main():
student_ids = range(1, 16)
semaphore = asyncio.Semaphore(5)
start_time = time.perf_counter()
tasks = [query_student_async(student_id, semaphore) for student_id in student_ids]
results = await asyncio.gather(*tasks)
end_time = time.perf_counter()
print(f"查询数量：{len(results)}")
print(f"观察到的最大同时查询数量：{max_running_count}")
print(f"总耗时：{end_time - start_time:.2f} 秒")
if __name__ == "__main__":
asyncio.run(main())运行命令：
uv run python c04_semaphore_limit_concurrency/02_semaphore_basic.py可能看到类似结果：
开始查询学生 1，当前正在查询：1
开始查询学生 2，当前正在查询：2
开始查询学生 3，当前正在查询：3
开始查询学生 4，当前正在查询：4
开始查询学生 5，当前正在查询：5
结束查询学生 1
结束查询学生 2
...
查询数量：15
观察到的最大同时查询数量：5
总耗时：1.50 秒运行逻辑说明：
semaphore = asyncio.Semaphore(5)表示同一时间最多允许 5 个任务进入受限区域。- 每个任务执行到
async with semaphore:时，都会尝试申请一个名额。 - 如果还有名额，任务进入查询区。
- 如果没有名额，任务会等前面的任务释放名额。
- 离开
async with后，名额会自动释放。 - 所以最大同时查询数量不会超过 5。
注意：限制要包住真正消耗资源的操作。
在这个例子中，真正的等待是：
await asyncio.sleep(0.5)真实项目中，这里可能换成数据库查询或外部接口请求。
3. 批量查询函数的并发限制
需求：把并发限制封装进批量查询函数里，调用时只需要传入学生 ID 和限制数量。
功能分析：
- 准备
student_data模拟学生数据。 - 编写
query_student_with_limit(student_id, semaphore)查询单个学生。 - 编写
query_students_with_limit(student_ids, limit=5)批量查询。 - 在批量查询函数中只创建一次共享的
semaphore。 - 使用
gather()收集所有结果。 - 对查询不到的学生返回
None，后续统一处理。
文件名：03_limited_batch.py
import asyncio
import time
student_data = {
student_id: {"id": student_id, "name": f"学生{student_id}", "age": 13 + student_id % 3}
for student_id in range(1, 31)
}
async def query_student_with_limit(student_id, semaphore):
async with semaphore:
print(f"正在查询学生 {student_id}")
await asyncio.sleep(0.2)
return student_data.get(student_id)
async def query_students_with_limit(student_ids, limit=5):
semaphore = asyncio.Semaphore(limit)
tasks = [
query_student_with_limit(student_id, semaphore)
for student_id in student_ids
]
results = await asyncio.gather(*tasks)
return results
async def main():
student_ids = range(1, 51)
start_time = time.perf_counter()
results = await query_students_with_limit(student_ids, limit=5)
end_time = time.perf_counter()
found_students = [student for student in results if student is not None]
missing_count = len(results) - len(found_students)
print(f"查询结果数量：{len(results)}")
print(f"查询成功数量：{len(found_students)}")
print(f"查询不到数量：{missing_count}")
print(f"总耗时：{end_time - start_time:.2f} 秒")
if __name__ == "__main__":
asyncio.run(main())运行命令：
uv run python c04_semaphore_limit_concurrency/03_limited_batch.py可能看到类似结果：
正在查询学生 1
正在查询学生 2
正在查询学生 3
正在查询学生 4
正在查询学生 5
...
查询结果数量：50
查询成功数量：30
查询不到数量：20
总耗时：2.01 秒运行逻辑说明：
query_students_with_limit()是批量查询函数。semaphore = asyncio.Semaphore(limit)在这个函数里只创建一次。- 所有单个查询任务共享同一个
semaphore。 - 这样才能限制“同一时间最多运行几个查询”。
- 如果在每个任务内部重新创建一个新的
Semaphore，每个任务都有自己的限制器，就达不到统一限制效果。 limit=5表示最多 5 个查询同时进入受限区域。student_data只有 1 到 30 号学生，所以 31 到 50 会返回None。- 最后用列表推导式过滤掉查询不到的学生。
注意：这类封装在真实项目中很常见。
比如批量访问外部 API 时，可以先设置一个保守的 limit，例如 5、10 或 20，再根据接口限制、数据库连接池大小和机器资源逐步调整。
这个演示固定传入 limit=5。最终案例会增加参数校验：限制值必须大于等于 1，否则 Semaphore(0) 没有可用名额，查询会一直等。
本例的限制器由 query_students_with_limit() 每次调用时创建，只在当前批次的查询之间共享。如果两个批次同时运行，且各自设置 limit=5，合计最多可能有 10 个查询进入受限区域。若多个批次需要共同遵守同一上限，就要共享同一个限制器；后续服务开发时再结合资源的生命周期组织它。
4. 带并发限制的学生批量查询
本章最终案例是一个普通 Python 脚本。
项目结构：
_code_async/c04_semaphore_limit_concurrency/
├── 01_unlimited_tasks.py
├── 02_semaphore_basic.py
├── 03_limited_batch.py
└── 04_limited_batch_query.py本章不需要额外安装依赖。
文件名：04_limited_batch_query.py
import asyncio
import time
student_data = {
student_id: {"id": student_id, "name": f"学生{student_id}", "age": 13 + student_id % 3}
for student_id in range(1, 31)
}
async def query_student_async(student_id, semaphore):
async with semaphore:
await asyncio.sleep(0.2)
return student_data.get(student_id)
def format_student(student):
if student is None:
return None
return f"ID：{student['id']}，姓名：{student['name']}，年龄：{student['age']}"
async def query_students_with_limit(student_ids, limit=5):
if limit < 1:
raise ValueError("limit 必须大于等于 1")
semaphore = asyncio.Semaphore(limit)
tasks = [
query_student_async(student_id, semaphore)
for student_id in student_ids
]
results = await asyncio.gather(*tasks)
return results
async def main():
student_ids = range(1, 51)
limit = 5
start_time = time.perf_counter()
results = await query_students_with_limit(student_ids, limit=limit)
end_time = time.perf_counter()
found_students = [student for student in results if student is not None]
missing_count = len(results) - len(found_students)
print("查询成功的学生：")
for student in found_students[:10]:
print(format_student(student))
if len(found_students) > 10:
print(f"还有 {len(found_students) - 10} 个学生未展示")
print()
print(f"本次查询 ID 数量：{len(results)}")
print(f"查询成功数量：{len(found_students)}")
print(f"查询不到数量：{missing_count}")
print(f"并发限制数量：{limit}")
print(f"总耗时：{end_time - start_time:.2f} 秒")
if __name__ == "__main__":
asyncio.run(main())启动命令：
uv run python c04_semaphore_limit_concurrency/04_limited_batch_query.py运行结果可能类似：
查询成功的学生：
ID：1，姓名：学生1，年龄：14
ID：2，姓名：学生2，年龄：15
ID：3，姓名：学生3，年龄：13
ID：4，姓名：学生4，年龄：14
ID：5，姓名：学生5，年龄：15
ID：6，姓名：学生6，年龄：13
ID：7，姓名：学生7，年龄：14
ID：8，姓名：学生8，年龄：15
ID：9，姓名：学生9，年龄：13
ID：10，姓名：学生10，年龄：14
还有 20 个学生未展示
本次查询 ID 数量：50
查询成功数量：30
查询不到数量：20
并发限制数量：5
总耗时：2.01 秒运行结果说明：
- 本次要查询 50 个学生 ID。
student_data里只有 1 到 30 号学生。- 所以查询成功数量是 30，查询不到数量是 20。
limit = 5表示同一时间最多只有 5 个查询进入受限区域。- 每个查询等待 0.2 秒，50 个查询分批执行，总耗时大约在 2 秒左右。
- 具体耗时会受电脑性能和运行环境影响，有轻微差异是正常的。
query_students_with_limit()会拒绝小于 1 的限制值，避免查询一直等待名额。
本章当前代码还不是 Web 项目。
它只是一个普通 Python 脚本，用来讲清楚批量异步查询中的并发数量控制。
并发限制错误排查
| 现象 | 原因 | 修复方式 |
|---|---|---|
设置了 limit=5，实际同时查询数仍超过 5 | 每个查询任务内部创建了自己的限制器，没有共享名额。 | 在当前批次开始时创建一次限制器，再传给所有查询任务。 |
| 名额申请正常，但实际 I/O 仍一起开始 | async with semaphore: 只包住准备工作，真正的查询在代码块外执行。 | 把实际查询等待放进受限区域。 |
| 创建了限制器，实际同时查询数却没有变化 | 查询代码没有通过 async with semaphore: 申请名额。 | 使用上下文管理器包住查询；仅创建限制器不会自动限制任务。 |
用 running_count 排查时，要让加 1、实际查询等待、减 1 保持在同一段测量范围内。计数器统计的是已经进入查询、尚未结束的任务，包括正在等待外部结果的任务，不是正在使用 CPU 的任务数量。本章示例没有异常或取消分支；相关清理规则留到第 5 章。
本章小结
本章最核心的是三件事：
- 异步可以同时等待多个任务，但外部资源的同时访问量需要控制。
asyncio.Semaphore(limit)可以限制同一时间进入某段异步代码的任务数量。async with semaphore:要包住真正消耗外部资源的操作。
本章仍一次性创建批量查询的协程对象。Semaphore 控制进入查询区的数量，不控制整批任务的总数量。
项目里可以先写成这样的结构：
async def query_students_with_limit(student_ids, limit=5):
semaphore = asyncio.Semaphore(limit)
tasks = [query_student_async(student_id, semaphore) for student_id in student_ids]
results = await asyncio.gather(*tasks)
return results练习
练习 1：在 02_semaphore_basic.py 中把 Semaphore(5) 分别改成 Semaphore(1)、Semaphore(3) 和 Semaphore(10)。观察最大同时查询数是否分别不超过 1、3、10，并比较总耗时。
练习 2：在 02_semaphore_basic.py 中，把查询函数内的 async with semaphore: 代码块替换为 async with semaphore: pass，并将原块内的计数、打印、等待和返回语句全部移到它后面，保持原顺序。这样查询前虽然申请过名额，却立即释放，实际查询不受限制。检查最大同时查询数是否变成 15，而不是 5；随后恢复原代码，确认上限重新为 5。
练习 3：在 04_limited_batch_query.py 中把查询范围改为 range(1, 101)。检查总数为 100、成功数仍为 30、查询不到为 70，耗时约为 100 ÷ 5 × 0.2 = 4 秒。
练习 4：在 02_semaphore_basic.py 的 query_student_async() 中，紧接 global 声明后加入 semaphore = asyncio.Semaphore(5)，让每个查询都使用独立的限制器。检查最大同时查询数是否变成 15；删除新增语句，恢复使用 main() 传入的共享限制器，确认上限重新为 5。