如何在Python asyncio中实现带权重的并发控制？-Python教程-PHP中文网

asyncio.Semaphore 不支持权重，需用 asyncio.Lock 和 asyncio.Condition 封装加权限流器；核心是维护已用配额、阻塞超限请求并唤醒等待者，注意 weight 校验、释放时机与统一权重口径。

如何在python asyncio中实现带权重的并发控制？

asyncio.Semaphore 无法直接支持权重，得自己封装

标准 asyncio.Semaphore 只认“请求数”，不认“资源消耗量”。比如一个任务占 3 单位带宽、另一个占 1 单位，用普通信号量只能按 1:1 排队，实际会浪费或超限。
真正要按权重调度，必须手动维护剩余配额，并在 acquire() / release() 时做加减和等待判断。

用 asyncio.Lock + 计数器实现加权限流器

核心是用一个 asyncio.Lock 保护共享状态（当前已占用配额），再配合 asyncio.Condition 唤醒等待者。比纯 Semaphore 多几行，但逻辑清晰、无竞态：

class WeightedLimiter:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self._used = 0
        self._lock = asyncio.Lock()
        self._cond = asyncio.Condition(self._lock)

    async def acquire(self, weight: int):
        async with self._cond:
            while self._used + weight > self.capacity:
                await self._cond.wait()
            self._used += weight

    async def release(self, weight: int):
        async with self._cond:
            self._used -= weight
            self._cond.notify_all()

常见错误：直接用 asyncio.Semaphore(1) 包裹加减操作——这只能串行化修改，但没解决“谁该被唤醒”的问题；漏掉 notify_all() 会导致后续协程永久挂起。

千问云

千问云是一款AI大模型工具，阿里云推出的全新MaaS模型服务平台。

下载

权重值必须为正整数，且不能超过总容量

使用时注意：
• weight 必须是 int，浮点权重会引发类型错误或精度问题
• 若某次 acquire(5) 而 capacity=4，该调用永远阻塞（不会抛异常）
• 实际业务中建议在 acquire 前加校验，比如 if weight > self.capacity: raise ValueError("weight exceeds capacity")
• 并发高时，频繁的 notify_all() 可能唤醒过多协程，造成“惊群”；若只允许一个协程继续，可改用 notify(1)

与 aiohttp 配合控制请求带宽时，权重应映射到真实资源消耗

例如下载大文件时按预估字节数设权重，小 API 调用按 QPS 折算：
• 一个 10MB 文件请求设 weight=10（单位：MB）
• 一个需 200ms 处理的接口设 weight=2（单位：0.1s）
关键点在于权重定义要统一口径，否则限流失去意义。别把“连接数”和“内存占用”混在一个 WeightedLimiter 实例里管。

最易忽略的是释放时机：必须确保 release() 在 finally 块中调用，哪怕任务中途取消或抛出未捕获异常——否则配额永久泄露，整个限流器最终卡死。