asyncio.Semaphore 如何与限流装饰器结合限并发

6次阅读

应使用 async with semaphore 而非手动 acquire()/release(),因其自动异常安全释放;装饰器须为异步函数并接收预创建的 Semaphore 实例,避免新建或闭包共享;fastapi 中推荐依赖注入替代装饰器。

asyncio.Semaphore 如何与限流装饰器结合限并发

限流装饰器里直接 await semaphore.acquire() 会出错

因为 acquire() 是协程函数,不能在同步装饰器里直接调用。常见错误是写成 semaphore.acquire()(漏掉 await),结果返回一个协程对象而非布尔值,后续逻辑崩掉;或者强行 await 在普通函数里,触发 RuntimeError: await outside async function

  • 装饰器本身必须是异步的(即返回 async def 包裹的协程),或用 functools.wraps + asyncio.create_task 做调度层包装
  • 信号量实例需在事件循环生命周期内共享,不能每次装饰都新建 asyncio.Semaphore(5),否则限流失效
  • 推荐把 semaphore 作为参数传入装饰器工厂,而非闭包捕获——避免多个装饰器实例误用同一变量引发竞态

用 async def 装饰器工厂实现可配置限流

这是最清晰、易测、可复用的方式。装饰器不自己管理信号量,而是接收一个已初始化的 semaphore 实例:

import asyncio from functools import wraps 

def with_semaphore(semaphore): def decorator(func): @wraps(func) async def wrapper(*args, *kwargs): async with semaphore: # 自动 acquire/release,含异常安全 return await func(args, **kwargs) return wrapper return decorator

全局信号量(按业务场景统一管理)

api_semaphore = asyncio.Semaphore(3)

@with_semaphore(api_semaphore) async def fetch_user(user_id: int): await asyncio.sleep(0.5) return {"id": user_id, "name": "alice"}

  • async with semaphore 是唯一推荐方式:它隐式调用 acquire()release(),即使 func 抛异常也保证释放
  • 若需区分不同接口并发上限(如 /users 限 3,/orders 限 10),就定义多个 semaphore 实例,分别传给对应装饰器
  • 别在装饰器里做 await asyncio.sleep() 或其他 I/O —— 它只负责准入控制,逻辑应下沉到被装饰函数中

FastAPI 路由中用依赖注入更自然

在 FastAPI 这类框架里,硬套装饰器反而绕路。直接用 Depends 注入带信号量的依赖,语义更正、测试更方便:

from fastapi import Depends, FastAPI import asyncio 

app = FastAPI() sem = asyncio.Semaphore(5)

async def rate_limited(): async with sem: yield

@app.get("/data") async def getdata( = Depends(rate_limited)): return {"status": "ok"}

  • 依赖函数 rate_limited 返回的是上下文管理器,Depends 会自动 await 并清理
  • 这种方式天然支持作用域scope="request"),且可与 BackgroundTasks中间件等正交组合
  • 如果需要动态限流(如按用户 Token 限频),就别用全局 semaphore,改用字典缓存 per-user 的 asyncio.Semaphore 实例,但要注意内存泄漏和过期清理

手动 acquire/release 的坑比你想象得多

有人为了“灵活控制”,放弃 async with,改用手动 acquire() + try/finally release()。这在简单场景看似可控,但极易出错:

  • 一旦 func 内部抛出未捕获异常,finally 可能来不及执行,导致信号量永久卡死(死锁)
  • 协程取消(asyncio.CancelledError)时,finally 不一定运行,release() 就丢了
  • acquire() 本身可能被取消 —— 它返回一个可等待对象,若在等待期间被 cancel,不会自动回滚计数器
  • 结论:除非你真需要在 acquire() 后、执行前插入日志或审计逻辑,否则一律用 async with。它不是语法糖,是安全边界。

信号量不是魔法,它只管“进来的数量”,不管“出去的速度”。如果你的协程平均耗时翻倍,而并发数没调低,排队延迟就会指数上升——这点常被忽略。

text=ZqhQzanResources