API限流的实现方法通常有两种:基于时间窗口的限流和令牌桶限流。
import time
# 定义一个字典用于存储每个IP的请求记录
requests = {}
def is_allowed(ip, limit, window):
now = time.time()
# 检查是否有该IP的请求记录,如果没有则新增一条
if ip not in requests:
requests[ip] = [(now, 1)]
return True
# 移除时间窗口外的请求记录
requests[ip] = [(t, c) for t, c in requests[ip] if now - t <= window]
# 统计时间窗口内的请求数
count = sum(c for _, c in requests[ip])
# 检查请求是否超过限制
if count < limit:
requests[ip].append((now, 1))
return True
return False
# 示例调用
ip = "127.0.0.1"
limit = 100 # 限制每秒最大请求数
window = 1 # 时间窗口为1秒
for i in range(110):
if is_allowed(ip, limit, window):
print("允许访问")
else:
print("请求过于频繁")
time.sleep(0.1)
import time
# 定义一个令牌桶类
class TokenBucket:
def __init__(self, capacity, rate):
self.capacity = capacity # 令牌桶的容量
self.tokens = capacity # 当前令牌数
self.rate = rate # 令牌生成速率
self.last_time = time.time() # 上一次生成令牌的时间
def get_token(self):
now = time.time()
# 计算距离上次生成令牌的时间间隔
duration = now - self.last_time
# 计算生成的新令牌数量
new_tokens = duration * self.rate
# 更新上次生成令牌的时间
self.last_time = now
# 令牌数不能超过容量限制
self.tokens = min(self.capacity, self.tokens + new_tokens)
# 检查是否有足够的令牌
if self.tokens >= 1:
self.tokens -= 1
return True
return False
# 示例调用
limit = 100 # 限制每秒最大请求数
rate = 10 # 令牌生成速率为10个/秒
bucket = TokenBucket(limit, rate)
for i in range(110):
if bucket.get_token():
print("允许访问")
else:
print("请求过于频繁")
time.sleep(0.1)
以上是两种常见的API限流实现方法,具体选择哪种方法取决于实际需求和系统情况。