在 Python 中，实现“限流”（Rate Limiting）通常是指限制某个操作在单位时间内可以执行的次数。如果未达到限流标准（比如请求频率过高），则阻塞或拒绝服务。 一种常见的限流算法是**令牌桶算法**（Token Bucket Algorithm）。下面是一个使用 `time` 模块实现的简单限流模块，当超过速率限制时会 **阻塞等待** 直到有新的令牌可用。 --- ### ✅ 示例：使用令牌桶实现的限流模块 ```python import time class RateLimiter: def __init__(self, rate, per): """ :param rate: 允许执行的次数 :param per: 单位时间（秒） """ self.rate = rate self.per = per self.tokens = rate # 当前可用令牌数 self.last_check = time.time() # 上次检查时间 def _renew_tokens(self): now = time.time() elapsed = now - self.last_check self.last_check = now # 根据流逝的时间增加令牌 self.tokens += elapsed * (self.rate / self.per) if self.tokens > self.rate: self.tokens = self.rate # 不超过最大容量 def acquire(self): """ 获取一个令牌，如果没有则阻塞等待。 """ while True: self._renew_tokens() if self.tokens >= 1: self.tokens -= 1 return else: # 等待一点时间再尝试 time.sleep(0.01) # 使用示例 if __name__ == "__main__": limiter = RateLimiter(rate=3, per=5) # 每5秒最多3次 for i in range(10): limiter.acquire() print(f"{i+1}: 执行了操作") ``` --- ### 🔍 解释： - `__init__`: 初始化限流器，每 `per` 秒允许 `rate` 次操作。 - `_renew_tokens`: 每次调用时根据时间差补充令牌，模拟令牌流入桶的过程。 - `acquire`: 尝试获取一个令牌，如果没有就休眠一小段时间继续尝试，直到成功为止。 - 示例中设置为每 5 秒允许最多 3 次操作，超出后程序会自动阻塞等待。 --- ###