asyncio.Semaphore必须全局共享，不能在装饰器内每次新建；正确做法是在模块级初始化后传入装饰器，使用async with确保协程安全，并优先选用BoundedSemaphore防止计数错乱。
asyncio.Semaphore 必须全局共享，不能在装饰器里每次新建
直接在装饰器内部 asyncio.Semaphore(5) 是常见错误——每次调用被装饰函数都会创建新实例，限流完全失效。信号量必须在模块级或类属性层面初始化一次，再透传或闭包捕获。
- ✅ 正确：在模块顶部定义
sem = asyncio.Semaphore(3)，装饰器内只引用它 - ❌ 错误：
def rate_limit(n=3): return lambda f: async def wrapper(...): sem = asyncio.Semaphore(n); async with sem: ...→ 每次 wrapper 都新建信号量 - 装饰器本身不感知事件循环，所以不能在
__call__里创建sem；必须提前构造好并绑定
写一个可复用的 async 限流装饰器，支持传参和协程安全
核心是让装饰器接收已创建的 semaphore 实例（而非数字），避免隐式状态。这样既灵活又可控，还能配合配置热重载。
import asynciodef with_semaphore(sem): def decorator(func): async def wrapper(*args, *kwargs): async with sem: return await func(args, **kwargs) return wrapper return decorator
使用示例
sem = asyncio.Semaphore(2) @with_semaphore(sem) async def fetch_data(url): await asyncio.sleep(0.5) # 模拟请求 return f"done: {url}"
- 不推荐用
@rate_limit(3)这种“数字即限流”的写法，容易掩盖信号量生命周期问题 - 若需动态调整上限，应替换
sem引用（如用asyncio.BoundedSemaphore+ 外部锁保护），而非改装饰器参数 - 装饰器返回的
wrapper必须是async def，否则await func()会报RuntimeWarning: coroutine 'xxx' was never awaited
与 FastAPI / aiohttp 等框架集成时，注意作用域和生命周期
在 Web 框架中，semaphore 应作为全局单例或依赖注入对象存在，绝不能在每个请求路径函数里 new 一个。
- FastAPI 中：在
main.py顶层定义sem = asyncio.Semaphore(10)，然后在路由函数或依赖中直接使用 - aiohttp 中：把
sem作为ClientSession的上下文属性或中间件状态传递，不要塞进request.app['sem']后每次取 —— 它本来就是线程/协程安全的，无需额外封装 - 如果用类封装业务逻辑（如
APIClient），把sem作为__init__参数传入并存为实例属性，比用@staticmethod+ 全局变量更清晰
为什么不用 asyncio.BoundedSemaphore？它更适合装饰器场景
asyncio.BoundedSemaphore 会在 release() 超出初始值时抛出 ValueError，这对装饰器尤其重要——万一异常导致 async with 未执行完，普通 Semaphore 会悄悄“多释放”，最终计数错乱、限流失效。
- 装饰器里若混用手动
acquire()/release()（比如加日志或超时逻辑），BoundedSemaphore是兜底保险 - 但只要坚持用
async with sem:，两者行为一致；差异只在“防御性编程”层面 - 初始化仍用相同参数：
asyncio.BoundedSemaphore(5)，其余代码无需改动
实际项目中最容易被忽略的一点：信号量不是“请求限流器”，它只控制同时进入某段代码的协程数。如果你在装饰器里限的是「数据库查询」，但被装饰函数里又并发发了 5 个 HTTP 请求，那这 5 个请求依然不受控——限流粒度必须和你要保护的资源严格对齐。
