asyncio.Semaphore与限流装饰器使用教程

golang学习网今天将给大家带来《asyncio.Semaphore 与限流装饰器结合使用，可以有效地控制并发数量，防止系统过载。以下是一个简单的实现方式：1. 使用 asyncio.Semaphore 控制并发asyncio.Semaphore 是一个异步信号量，用于限制同时运行的协程数量。import asyncio semaphore = asyncio.Semaphore(5) # 允许最多5个并发任务 async def limited_task(task_id): async with semaphore: print(f"Task {task_id} is running") await asyncio.sleep(1) print(f"Task {task_id} completed")2. 创建限流装饰器我们可以创建一个装饰器，在函数调用前获取信号量，从而限制并发。def rate_limit(max_concurrent=5): semaphore = asyncio.Semaphore(max_concurrent) def decorator(func): async def wrapper(*args, **kwargs): async with semaphore: return await func(*args, **kwargs) return wrapper return decorator3. 应用装饰器到异步函数@rate_limit(max_concurrent=3) async def my_async_function(param): print(f"Processing {param}") await asyncio.sleep(1) print(f"Finished {param}")4. 启动多个任务》，感兴趣的朋友请继续看下去吧！以下内容将会涉及到等等知识点，如果你是正在学习文章或者已经是大佬级别了，都非常欢迎也希望大家都能给我建议评论哈~希望能帮助到大家！

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

限流装饰器里直接 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
<p>def with_semaphore(semaphore):
def decorator(func):
@wraps(func)
async def wrapper(*args, *<em>kwargs):
async with semaphore:  # 自动 acquire/release，含异常安全
return await func(</em>args, **kwargs)
return wrapper
return decorator</p><h1>全局信号量（按业务场景统一管理）</h1><p>api_semaphore = asyncio.Semaphore(3)</p><p>@with_semaphore(api_semaphore)
async def fetch_user(user_id: int):
await asyncio.sleep(0.5)
return {"id": user_id, "name": "alice"}</p>

• 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
<p>app = FastAPI()
sem = asyncio.Semaphore(5)</p><p>async def rate_limited():
async with sem:
yield</p><p>@app.get("/data")
async def get<em>data(</em> = Depends(rate_limited)):
return {"status": "ok"}</p>

• 依赖函数 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。它不是语法糖，是安全边界。

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

今天关于《asyncio.Semaphore与限流装饰器使用教程》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！