asyncio.Semaphore 是协程级计数信号量，用于限制并发数，需用 async with 确保 acquire/release 配对；必须全局复用、在临界区（如HTTP请求）内使用，而非包裹 create_task。
asyncio.Semaphore 是什么，为什么不能直接用 asyncio.create_task
它不是“限制任务数量”的开关，而是协程级别的计数信号量——每次 acquire() 消耗一个许可，release() 归还一个。不手动配对或没用 async with，许可就卡住，后续任务永远等下去。
常见错误现象：asyncio.TimeoutError、大量任务卡在 await semaphore.acquire()、CPU 占用低但响应停滞。
- 必须用
async with semaphore:包裹临界区，否则release()容易被异常跳过 - 不要在
create_task外层套async with，那会把整个 task 创建过程锁住，失去并发意义 - 信号量对象要复用，不能每个任务都新建一个
asyncio.Semaphore(3)
怎么写才真正限制住并发数（带示例）
核心是：把需要限流的 I/O 操作（比如 HTTP 请求、数据库查询）放进 async with semaphore: 块里，而不是包裹 create_task 调用本身。
import asyncio
import aiohttp
<p>semaphore = asyncio.Semaphore(3) # 全局一个，最多 3 个并发</p><div class="aritcle_card flexRow">
<div class="artcardd flexRow">
<a class="aritcle_card_img" href="/xiazai/skill9965" title="macOS UI 自动化技能包 —— 用 Python 控制你的 Mac"><img
src="https://img.php.cn/upload/skill/001/246/273/6adaf20dbae7c507.jpg" alt="macOS UI 自动化技能包 —— 用 Python 控制你的 Mac" ></a>
<div class="aritcle_card_info flexColumn">
<a href="/xiazai/skill9965" title="macOS UI 自动化技能包 —— 用 Python 控制你的 Mac">macOS UI 自动化技能包 —— 用 Python 控制你的 Mac</a>
<p>让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript
系统要求
macOS 14+ (Sonoma)
Python 3.9+
需要开启：辅助功能 + 屏幕录制权限</p>
</div>
<a href="/xiazai/skill9965" title="macOS UI 自动化技能包 —— 用 Python 控制你的 Mac" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span> </a>
</div>
</div><p><span>立即学习</span>“<a href="https://pan.quark.cn/s/00968c3c2c15" style="text-decoration: underline !important; color: blue; font-weight: bolder;" rel="nofollow" target="_blank">Python免费学习笔记（深入）</a>”；</p><p>async def fetch(url):
async with semaphore: # ✅ 这里才是关键入口
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return await resp.text()</p><h1>启动 10 个任务，但只有 3 个能同时发请求</h1><p>tasks = [fetch(f"<a href="https://www.php.cn/link/7894fb70687d7d6ed620d74d67420eb5">https://www.php.cn/link/7894fb70687d7d6ed620d74d67420eb5</a>}") for i in range(10)]
results = await asyncio.gather(*tasks)-
semaphore必须定义在所有任务共享的作用域（如模块级、类属性），不能在函数内反复创建 - 如果任务内部还有子协程也要限流，得传入同一个
semaphore实例 - 别用
loop.run_until_complete手动管理，直接asyncio.run(main())更安全
和 asyncio.BoundedSemaphore 有什么区别
绝大多数场景下，用 asyncio.Semaphore 就够了。BoundedSemaphore 会在 release() 超出初始值时抛 ValueError，但它只在你手动调用 acquire/release 且逻辑混乱时才有意义。
- 用
async with就不会触发BoundedSemaphore的边界检查，两者行为一致 - 除非你在异常处理里反复
release又不acquire，否则没必要换 - 标准库文档也明确说：BoundedSemaphore 主要用于调试信号量使用是否匹配
容易被忽略的坑：超时 + 限流组合下的死锁风险
当限流和 asyncio.wait_for 一起用，如果超时发生在 acquire() 阶段，信号量不会自动释放——因为还没进入 async with 块。
- 避免在
wait_for里包整个async with semaphore:，而应先await asyncio.wait_for(semaphore.acquire(), timeout=5)，再进async with - 更稳妥的做法：用
asyncio.timeout（Python 3.11+）替代wait_for，它能正确清理未完成的acquire - 如果用了旧版本 Python，务必给
acquire()单独加超时，并在except里确认没漏掉release
信号量本身不记录谁占了许可，出错后靠人盯日志很难定位哪次 acquire 没配对——所以宁可多写两行超时控制，别省那点代码。
