### Python 中实现速率限制或节流机制的方法 为了在 Python 中限制函数调用的请求频率，可以通过引入速率限制（Rate Limiting）或节流（Throttling）机制来控制资源访问的速度。这通常用于网络请求、API 调用或其他高并发场景下保护服务器免受过载影响。 一种常见的方法是通过令牌桶算法（Token Bucket Algorithm）。该算法的核心思想是一个固定容量的桶会按照一定速率填充令牌，每次执行操作前需要消耗一个令牌。如果桶中的令牌不足，则阻止进一步的操作直至有足够的令牌可用[^1]。 以下是基于 `threading` 和自定义类的一个简单实现： #### 使用 TokenBucket 类实现速率限制 ```python import time from threading import Lock class TokenBucket: def __init__(self, rate_per_second): self.rate = rate_per_second # 每秒生成多少个令牌 self.capacity = rate_per_second # 容量等于每秒最大生成数量 self.tokens = self.capacity # 初始满额 self.last_consume_time = time.time() self.lock = Lock() def consume(self, amount=1): with self.lock: current_time = time.time() elapsed_time = current_time - self.last_consume_time new_tokens = elapsed_time * self.rate if new_tokens > 0: self.tokens = min(self.capacity, self.tokens + new_tokens) self.last_consume_time = current_time if self.tokens >= amount: self.tokens -= amount return True else: return False ``` 在这个例子中，`consume()` 方法尝试获取指定数量的令牌。如果没有足够的令牌可供消费，则返回 `False` 表明当前无法继续操作；否则减少相应数量的令牌并允许操作继续进行。 接着可以在实际业务逻辑里应用这个限流器： #### 应用到具体的工作流程中 ```python def throttled_function(throttle_instance, func_to_call, *args, **kwargs): while not throttle_instance.consume(): time.sleep(0.1) # 如果没有足够令牌则短暂休眠再试 return func_to_call(*args, **kwargs) # 假设我们有一个 API 请求函数 def api_request(url): print(f"Making request to {url}") # 实际发送 HTTP 请求... throttle = TokenBucket(rate_per_second=2) # 设置为每秒最多两次请求 for i in range(5): url = f"http://example.com/api/{i}" result = throttled_function(throttle, api_request, url) ``` 此代码片段展示了如何利用前面定义好的 `TokenBucket` 来约束对某个假想 API 接口发起的一系列连续请求行为。它确保不会超过设定的最大 QPS (Queries Per Second)。 另外还有现成库可以帮助简化这一过程比如 [`limits`](https://github.com/yohanboniface/limits)，或者框架自带的支持如 Flask-Limiter 对于 Web 开发者来说非常方便实用。 ---