Python异步开发中的信号量控制_使用asyncio.Semaphore限制资源访问-Python教程-PHP中文网

asyncio.Semaphore 不能直接 await，因其非协程对象，需通过 acquire()（协程）和 release()（普通方法）配合使用，推荐用 async with semaphore: 确保安全获取与释放。

python异步开发中的信号量控制_使用asyncio.semaphore限制资源访问

asyncio.Semaphore 为什么不能直接 await？

因为 asyncio.Semaphore 本身不是协程，await semaphore 会报 TypeError: object Semaphore can't be used in 'await' expression。它得通过 acquire() 和 release() 配合使用——前者是协程，后者是普通方法（但必须由同一线程/协程调用）。

常见错误是写成：

await semaphore  # ❌ 错误！

正确姿势是：

用 async with semaphore: 最安全，自动处理获取和释放
手动 await semaphore.acquire() + semaphore.release()，但必须确保 release() 一定执行（比如套 try/finally）
别在不同协程间传递未释放的 semaphore 实例，容易漏 release 导致死锁

并发数限制不准？检查是否复用了同一个 Semaphore 实例

如果你期望最多 3 个并发请求，但实际跑出了 5 个，大概率是多个协程共享了同一个 semaphore，但没意识到 acquire() 是可重入的——它只计数，不绑定调用者。问题常出在：把 semaphore 定义在模块顶层、或作为类属性，却被多个异步任务反复 await。

典型场景：

Web 请求批量调用中，所有 fetch_url() 共享一个全局 semaphore ✅
但每个 fetch_url(url) 内部又自己新建 semaphore = asyncio.Semaphore(3) ❌ —— 每次都新建，完全失效
或在类方法里写 self.semaphore = asyncio.Semaphore(3) 却没做单例控制，导致每实例化一次就多一套计数器

验证方式：打印 semaphore._value（非公开属性，仅调试用），看是否随并发上升而递减。

为什么 await semaphore.acquire() 有时卡住不动？

最常见原因是：某个协程拿到信号量后崩溃、提前 return、或忘了 release()，导致计数器卡死。这时后续所有 acquire() 都会挂起，直到有 release() 被调用。

python-testing-patterns

使用 pytest、fixtures、mock 和测试驱动开发实现全面的测试策略。用于编写 Python 测试、搭建测试套件或实施测试最佳实践。

下载

避免方法：

强制用 async with semaphore:，Python 3.7+ 支持，底层自动处理异常路径下的释放
不用 async with 时，必须包 try/finally：

await semaphore.acquire()
try:
    await do_something()
finally:
    semaphore.release()

注意：semaphore.release() 不是协程，不能 await；且不能对已释放的 semaphore 多次 release（会抛 ValueError）。

和线程版 threading.Semaphore 混用会怎样？

完全不可互换。asyncio.Semaphore 是为事件循环设计的，内部依赖 asyncio.get_event_loop()；threading.Semaphore 是阻塞式，一旦在协程里 semaphore.acquire()（线程版），就会阻塞整个 event loop，拖垮所有并发任务。

真实踩坑点：