本文目录导读：
我来给你几个不同场景的滑动窗口限流案例,从简单到复杂。
基础版：固定窗口计数器（演示问题）
import time
import threading
class FixedWindowRateLimiter:
"""固定窗口限流器 - 演示边界问题"""
def __init__(self, max_requests, window_seconds):
self.max_requests = max_requests # 窗口内最大请求数
self.window_seconds = window_seconds # 窗口大小
self.current_window_start = time.time()
self.request_count = 0
self.lock = threading.Lock()
def allow_request(self):
with self.lock:
now = time.time()
# 如果当前时间超出窗口，重置计数
if now - self.current_window_start >= self.window_seconds:
self.current_window_start = now
self.request_count = 0
# 检查是否超限
if self.request_count < self.max_requests:
self.request_count += 1
return True
return False
# 演示边界问题
limiter = FixedWindowRateLimiter(max_requests=5, window_seconds=60)
# 在窗口边界处可能绕过限流
滑动窗口日志版（精确但耗内存）
from collections import deque
import time
import threading
class SlidingWindowLogRateLimiter:
"""滑动窗口日志限流器 - 使用时间戳记录"""
def __init__(self, max_requests, window_seconds):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = deque() # 存储请求时间戳
self.lock = threading.Lock()
def allow_request(self):
with self.lock:
now = time.time()
# 移除窗口外的旧请求
while self.requests and now - self.requests[0] >= self.window_seconds:
self.requests.popleft()
# 检查当前窗口内请求数
if len(self.requests) < self.max_requests:
self.requests.append(now)
return True
return False
def get_current_count(self):
"""获取当前窗口内请求数"""
with self.lock:
now = time.time()
while self.requests and now - self.requests[0] >= self.window_seconds:
self.requests.popleft()
return len(self.requests)
# 使用示例
limiter = SlidingWindowLogRateLimiter(max_requests=10, window_seconds=60)
for i in range(15):
if limiter.allow_request():
print(f"请求 {i+1}: 允许")
else:
print(f"请求 {i+1}: 拒绝")
滑动窗口计数器版（高效实用）
import time
import threading
from collections import deque
class SlidingWindowCounterRateLimiter:
"""滑动窗口计数器限流器 - 使用子窗口聚合"""
def __init__(self, max_requests, window_seconds, sub_window_count=10):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.sub_window_size = window_seconds / sub_window_count # 每个子窗口大小
self.windows = deque() # (window_start_time, count) 对
self.total_count = 0
self.lock = threading.Lock()
def allow_request(self):
with self.lock:
now = time.time()
self._cleanup(now)
if self.total_count < self.max_requests:
self._add_request(now)
return True
return False
def _cleanup(self, now):
"""清理过期的子窗口"""
while self.windows and now - self.windows[0][0] >= self.window_seconds:
_, count = self.windows.popleft()
self.total_count -= count
def _add_request(self, now):
"""添加请求到当前子窗口"""
# 找到当前子窗口的起始时间
current_window_start = now - (now % self.sub_window_size)
# 如果当前子窗口不存在或已过期，创建新的
if not self.windows or self.windows[-1][0] < current_window_start:
self.windows.append((current_window_start, 1))
else:
# 更新当前子窗口的计数
window_start, count = self.windows[-1]
self.windows[-1] = (window_start, count + 1)
self.total_count += 1
def get_stats(self):
"""获取统计信息"""
with self.lock:
now = time.time()
self._cleanup(now)
return {
'total_requests': self.total_count,
'window_size': self.window_seconds,
'sub_windows': [(time.strftime('%H:%M:%S', time.localtime(ws)), c)
for ws, c in self.windows]
}
# 使用示例
limiter = SlidingWindowCounterRateLimiter(max_requests=100, window_seconds=60, sub_window_count=6)
Redis实现（分布式场景）
import redis
import time
import json
class RedisSlidingWindowRateLimiter:
"""使用Redis实现分布式滑动窗口限流"""
def __init__(self, redis_client, key_prefix='ratelimit', max_requests=100, window_seconds=60):
self.redis = redis_client
self.key_prefix = key_prefix
self.max_requests = max_requests
self.window_seconds = window_seconds
def allow_request(self, user_id):
"""
使用Redis ZSET实现滑动窗口
每个元素为请求时间戳，score为时间戳
"""
key = f"{self.key_prefix}:{user_id}"
now = time.time()
# 使用pipeline保证原子性
pipe = self.redis.pipeline()
# 添加当前请求
pipe.zadd(key, {str(now): now})
# 移除窗口外的旧请求
pipe.zremrangebyscore(key, 0, now - self.window_seconds)
# 获取当前窗口内请求数
pipe.zcard(key)
# 设置过期时间
pipe.expire(key, self.window_seconds)
_, _, request_count, _ = pipe.execute()
if request_count <= self.max_requests:
return True
return False
def get_remaining(self, user_id):
"""获取剩余可用请求数"""
key = f"{self.key_prefix}:{user_id}"
now = time.time()
# 清理过期数据
self.redis.zremrangebyscore(key, 0, now - self.window_seconds)
current_count = self.redis.zcard(key)
return max(0, self.max_requests - current_count)
# 分布式使用示例
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
limiter = RedisSlidingWindowRateLimiter(
redis_client,
max_requests=100,
window_seconds=60
)
完整实战案例：API网关限流
from fastapi import FastAPI, HTTPException, Request
from typing import Optional
import time
import threading
from collections import defaultdict, deque
from datetime import datetime
import asyncio
class APIRateLimiter:
"""完整API限流器，支持多用户、多接口"""
def __init__(self):
self.limits = {
'default': {'max_requests': 100, 'window': 60},
'user': {'max_requests': 10, 'window': 60},
'path': {'max_requests': 30, 'window': 60},
'admin': {'max_requests': 1000, 'window': 60},
}
self.user_records = defaultdict(deque)
self.path_records = defaultdict(deque)
self.ip_records = defaultdict(deque)
self.lock = threading.Lock()
self.blocked_ips = {}
def allow_request(self, user_id: Optional[str] = None,
path: str = None, ip: str = None) -> dict:
"""
检查请求是否允许
返回：{'allowed': bool, 'retry_after': seconds, 'remaining': int}
"""
now = time.time()
with self.lock:
# 检查IP是否被临时封禁
if ip in self.blocked_ips:
block_until = self.blocked_ips[ip]
if now < block_until:
return {
'allowed': False,
'retry_after': int(block_until - now),
'remaining': 0,
'reason': 'IP暂时被封禁'
}
else:
del self.blocked_ips[ip]
# 检查各类限制
checks = []
# 用户限制
if user_id:
check = self._check_limit(
self.user_records[user_id],
self.limits['user']['max_requests'],
self.limits['user']['window'],
now
)
checks.append(check)
# 路径限制
if path:
check = self._check_limit(
self.path_records[path],
self.limits['path']['max_requests'],
self.limits['path']['window'],
now
)
checks.append(check)
# IP限制
if ip:
check = self._check_limit(
self.ip_records[ip],
self.limits['default']['max_requests'],
self.limits['default']['window'],
now
)
checks.append(check)
# 汇总结果
if checks and all(c['allowed'] for c in checks):
return {
'allowed': True,
'remaining': min(c['remaining'] for c in checks),
'retry_after': 0
}
else:
# 找到最长的等待时间
retry_after = max(c['retry_after'] for c in checks if c['retry_after'] > 0)
# 连续失败可能触发临时封禁
if checks and any(c['remaining'] == 0 for c in checks) and ip:
self._block_ip(ip, now)
return {
'allowed': False,
'retry_after': retry_after,
'remaining': 0,
'reason': '达到限流阈值'
}
def _check_limit(self, records: deque, max_requests: int,
window: int, now: float) -> dict:
"""检查单个限流规则"""
# 清理过期记录
while records and now - records[0] >= window:
records.popleft()
remaining = max_requests - len(records)
if remaining > 0:
records.append(now)
return {'allowed': True, 'remaining': remaining, 'retry_after': 0}
else:
oldest = records[0] if records else now
retry_after = max(0, window - (now - oldest))
return {'allowed': False, 'remaining': 0, 'retry_after': retry_after}
def _block_ip(self, ip: str, now: float, duration: int = 300):
"""临时封禁IP"""
self.blocked_ips[ip] = now + duration
# FastAPI集成示例
app = FastAPI()
limiter = APIRateLimiter()
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
"""全局限流中间件"""
user_id = request.headers.get('X-User-ID')
ip = request.client.host
path = request.url.path
result = limiter.allow_request(
user_id=user_id,
path=path,
ip=ip
)
if not result['allowed']:
raise HTTPException(
status_code=429,
detail=f"请求过于频繁，请在 {result['retry_after']} 秒后重试",
headers={
'Retry-After': str(result['retry_after']),
'X-RateLimit-Remaining': str(result['remaining'])
}
)
response = await call_next(request)
# 添加限流响应头
response.headers['X-RateLimit-Remaining'] = str(result['remaining'])
return response
# 示例接口
@app.get("/api/data")
async def get_data(request: Request):
return {"message": "成功获取数据"}
# 测试代码
def test_rate_limiter():
"""测试限流器"""
limiter = APIRateLimiter()
# 测试用户限流
print("=== 用户限流测试 ===")
for i in range(15):
result = limiter.allow_request(user_id="user1", path="/api/data", ip="192.168.1.1")
if result['allowed']:
print(f"请求 {i+1}: ✓ 允许 (剩余: {result['remaining']})")
else:
print(f"请求 {i+1}: ✗ 拒绝 (等待: {result['retry_after']}秒, {result.get('reason', '')})")
# 模拟等待
print("\n=== 等待5秒后 ===")
time.sleep(5)
# 测试管理员
print("\n=== 管理员限流测试 ===")
for i in range(5):
result = limiter.allow_request(user_id="admin", path="/api/data", ip="192.168.1.2")
print(f"请求 {i+1}: {'✓ 允许' if result['allowed'] else '✗ 拒绝'}")
if __name__ == "__main__":
test_rate_limiter()
高级优化版本：自适应限流
import threading
import time
from collections import deque
from typing import Dict, Tuple
class AdaptiveSlidingWindowLimiter:
"""自适应滑动窗口限流器"""
def __init__(self, max_requests, window_seconds,
threshold_high=0.8, threshold_low=0.3):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.threshold_high = threshold_high # 触发扩缩容的高阈值
self.threshold_low = threshold_low # 触发扩容的低阈值
self.windows: Dict[int, int] = {} # {子窗口起始时间: 请求数}
self.total_count = 0
self.lock = threading.Lock()
# 动态调整参数
self.current_max = max_requests
self.adjustment_counter = 0
def allow_request(self) -> bool:
with self.lock:
now = time.time()
self._cleanup(now)
if self.total_count < self.current_max:
self._add_request(now)
self._adjust_capacity(now)
return True
return False
def _cleanup(self, now: float):
"""清理过期的子窗口"""
expire_threshold = now - self.window_seconds
for start_time in list(self.windows.keys()):
if start_time < expire_threshold:
self.total_count -= self.windows.pop(start_time)
def _add_request(self, now: float):
"""添加请求"""
window_start = int(now // self.window_seconds)
self.windows[window_start] = self.windows.get(window_start, 0) + 1
self.total_count += 1
def _adjust_capacity(self, now: float):
"""根据负载动态调整容量"""
self.adjustment_counter += 1
if self.adjustment_counter % 10 != 0: # 每10个请求评估一次
return
usage_ratio = self.total_count / self.current_max
if usage_ratio > self.threshold_high:
# 高负载，降低限制
self.current_max = max(1, int(self.current_max * 0.8))
elif usage_ratio < self.threshold_low:
# 低负载，提高限制
self.current_max = min(self.max_requests * 2,
int(self.current_max * 1.2))
def get_stats(self) -> Dict:
"""获取统计信息"""
with self.lock:
return {
'current_max': self.current_max,
'total_count': self.total_count,
'usage_percentage': (self.total_count / self.current_max * 100)
if self.current_max else 0,
'windows': len(self.windows)
}
# 测试自适应限流
def test_adaptive_limiter():
limiter = AdaptiveSlidingWindowLimiter(
max_requests=100,
window_seconds=60,
threshold_high=0.8,
threshold_low=0.3
)
print("=== 自适应限流测试 ===")
for i in range(200):
allowed = limiter.allow_request()
if i % 20 == 0:
stats = limiter.get_stats()
print(f"请求 {i:3d}: {'✓' if allowed else '✗'} | 当前容量: {stats['current_max']:4d} | "
f"已使用: {stats['total_count']:3d} | 使用率: {stats['usage_percentage']:.1f}%")
if __name__ == "__main__":
test_adaptive_limiter()
- 固定窗口问题：存在边界突发，可以通过滑动窗口解决
- 日志方式精确：记录每个请求的时间戳，但内存消耗大
- 计数器方式高效：通过子窗口聚合，权衡内存和精度
- Redis方案适合分布式：保证一致性，支持多实例部署
- 实际应用建议：
- 使用子窗口（如10个小窗口）平衡精度和性能
- 设置合理的缓存过期时间
- 支持动态调整参数
- 记录限流日志用于分析
这些案例覆盖了从简单到复杂的场景,你可以根据实际需求选择适合的实现方式。
