当前位置：首页 > 文章列表 > 文章 > python教程 > asyncio.Semaphore与限流装饰器使用教程

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

2026-02-04 15:52:45 0浏览收藏

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 中推荐依赖注入替代装饰器。

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
<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 实例，但要注意内存泄漏和过期清理