如何基于Python asyncio实现符合QPS限制的远程调用速率控制

asyncio没有直接内置QPS限流工具，但可以通过asyncio.Lock结合滑动时间窗口实现高效、优雅的限流，完全避免固定间隔sleep的低效问题。

核心实现 #

import asyncio
from collections import deque
from time import time

class QPSLimiter:
    def __init__(self, max_qps: int):
        self.max_qps = max_qps
        # 存储最近1秒内的调用时间戳
        self.call_timestamps = deque()
        # 协程安全锁，保证计数准确
        self._lock = asyncio.Lock()

    async def acquire(self):
        async with self._lock:
            now = time()
            # 清理1秒窗口外的旧调用记录
            while self.call_timestamps and now - self.call_timestamps[0] > 1.0:
                self.call_timestamps.popleft()
            
            # 如果当前窗口内调用数达上限，等待到最早的调用记录过期
            while len(self.call_timestamps) >= self.max_qps:
                wait_until = self.call_timestamps[0] + 1.0
                await asyncio.sleep(wait_until - now)
                now = time()
                # 再次清理过期记录
                while self.call_timestamps and now - self.call_timestamps[0] > 1.0:
                    self.call_timestamps.popleft()
            
            # 记录当前调用时间
            self.call_timestamps.append(now)

    # 支持上下文管理器语法，使用更简洁
    async def __aenter__(self):
        await self.acquire()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        pass

# 给业务函数添加限流包装
async def limited_fun(x: int, limiter: QPSLimiter) -> int:
    async with limiter:
        return await fun(x)

# 批量调用示例
async def main():
    limiter = QPSLimiter(max_qps=2)
    tasks = [limited_fun(i, limiter) for i in range(1000)]
    results = await asyncio.gather(*tasks)
    print(results)

if __name__ == "__main__":
    asyncio.run(main())

方案优势 #

• 精准限流：滑动时间窗口严格控制每秒调用次数，不会出现瞬间超量；
• 高效异步：远程调用的20秒等待时间会被用来调度其他协程，完全利用异步优势；
• Pythonic：用上下文管理器封装限流逻辑，业务代码侵入性低，可读性强；
• 智能等待：仅在调用数达上限时才等待，不会像固定间隔sleep那样浪费时间。

注意事项 #

• 该实现是协程安全的，通过asyncio.Lock保证多协程环境下的计数正确性；
• 若需多接口独立限流，可为每个接口单独实例化QPSLimiter；
• 相比固定间隔方案，滑动窗口能更好应对突发请求，同时严格遵守QPS限制。

内容的提问来源于stack exchange，提问作者maplemaple