Python API限流策略：保护你的服务免受滥用

Python API限流策略：保护你的服务免受滥用

引言

在高并发的API服务中，限流是保障系统稳定性的关键措施。当请求量超过系统处理能力时，限流可以有效保护服务不被压垮，确保公平使用。作为一名从Python转向Rust的后端开发者，我在实践中总结了多种限流策略的实现方式。本文将深入探讨Python中API限流的设计与实现，帮助你构建健壮的限流系统。

一、限流核心概念

1.1 什么是限流

限流（Rate Limiting）是一种控制系统资源使用的技术，通过限制单位时间内的请求数量来保护服务。

1.2 限流的作用

• 防止服务过载：保护后端服务不被大量请求压垮
• 公平使用：确保每个用户都能公平访问资源
• 防止恶意攻击：抵御DoS/DDoS攻击
• 资源分配：合理分配系统资源

1.3 常见限流算法对比

二、限流算法实现

2.1 固定窗口计数器

import time
from collections import defaultdict

class FixedWindowCounter:
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.counters = defaultdict(int)
self.windows = {}

def is_allowed(self, key: str) -> bool:
now = time.time()
window_key = int(now // self.window_seconds)

if self.windows.get(key) != window_key:
self.counters[key] = 0
self.windows[key] = window_key

if self.counters[key] < self.max_requests:
self.counters[key] += 1
return True

return False

def get_remaining(self, key: str) -> int:
now = time.time()
window_key = int(now // self.window_seconds)

if self.windows.get(key) != window_key:
return self.max_requests

return self.max_requests – self.counters[key]

2.2 滑动窗口计数器

from collections import deque

class SlidingWindowCounter:
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.timestamps = defaultdict(deque)

def is_allowed(self, key: str) -> bool:
now = time.time()
window_start = now – self.window_seconds

timestamps = self.timestamps[key]

while timestamps and timestamps[0] < window_start:
timestamps.popleft()

if len(timestamps) < self.max_requests:
timestamps.append(now)
return True

return False

def get_remaining(self, key: str) -> int:
now = time.time()
window_start = now – self.window_seconds

timestamps = self.timestamps[key]
while timestamps and timestamps[0] < window_start:
timestamps.popleft()

return self.max_requests – len(timestamps)

2.3 令牌桶算法

class TokenBucket:
def __init__(self, capacity: int, rate: float):
self.capacity = capacity
self.rate = rate
self.tokens = capacity
self.last_refill = time.time()

def _refill(self):
now = time.time()
elapsed = now – self.last_refill
new_tokens = elapsed * self.rate

self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now

def is_allowed(self, tokens: int = 1) -> bool:
self._refill()

if self.tokens >= tokens:
self.tokens -= tokens
return True

return False

def get_remaining(self) -> float:
self._refill()
return self.tokens

2.4 漏桶算法

class LeakyBucket:
def __init__(self, capacity: int, rate: float):
self.capacity = capacity
self.rate = rate
self.queue = deque()
self.last_leak = time.time()

def _leak(self):
now = time.time()
elapsed = now – self.last_leak

leaks = int(elapsed * self.rate)
for _ in range(leaks):
if self.queue:
self.queue.popleft()

self.last_leak = now

def is_allowed(self) -> bool:
self._leak()

if len(self.queue) < self.capacity:
self.queue.append(time.time())
return True

return False

def get_remaining(self) -> int:
self._leak()
return self.capacity – len(self.queue)

三、分布式限流实现

3.1 基于Redis的分布式限流

import redis
import json

class RedisRateLimiter:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client

def fixed_window_limit(self, key: str, max_requests: int, window_seconds: int) -> bool:
window_key = f"rate_limit:{key}:{int(time.time() // window_seconds)}"

count = self.redis_client.incr(window_key)
if count == 1:
self.redis_client.expire(window_key, window_seconds)

return count <= max_requests

def sliding_window_limit(self, key: str, max_requests: int, window_seconds: int) -> bool:
now = time.time()
window_start = now – window_seconds

pipeline = self.redis_client.pipeline()
pipeline.zadd(f"rate_limit:{key}", {now: now})
pipeline.zremrangebyscore(f"rate_limit:{key}", 0, window_start)
pipeline.zcard(f"rate_limit:{key}")
pipeline.expire(f"rate_limit:{key}", window_seconds)

results = pipeline.execute()
count = results[2]

return count <= max_requests

def token_bucket_limit(self, key: str, capacity: int, rate: float) -> bool:
now = time.time()
token_key = f"token_bucket:{key}"

pipeline = self.redis_client.pipeline()
pipeline.get(token_key)

results = pipeline.execute()
stored = results[0]

if stored:
data = json.loads(stored)
last_refill = data['last_refill']
tokens = data['tokens']

elapsed = now – last_refill
new_tokens = elapsed * rate
tokens = min(capacity, tokens + new_tokens)
else:
tokens = capacity
last_refill = now

if tokens >= 1:
tokens -= 1
self.redis_client.set(token_key, json.dumps({
'tokens': tokens,
'last_refill': now
}))
return True

return False

四、限流中间件实现

4.1 Flask限流中间件

from flask import Flask, request, jsonify
from functools import wraps

app = Flask(__name__)
rate_limiter = RedisRateLimiter(redis.Redis())

def rate_limit(max_requests: int, window_seconds: int):
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
client_id = request.remote_addr

if not rate_limiter.fixed_window_limit(client_id, max_requests, window_seconds):
return jsonify({
'error': 'Rate limit exceeded',
'retry_after': window_seconds
}), 429

return f(*args, **kwargs)
return decorated_function
return decorator

@app.route('/api/data')
@rate_limit(max_requests=100, window_seconds=60)
def get_data():
return jsonify({'data': 'Hello, World!'})

4.2 FastAPI限流中间件

from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.base import BaseHTTPMiddleware

app = FastAPI()
rate_limiter = RedisRateLimiter(redis.Redis())

class RateLimitMiddleware(BaseHTTPMiddleware):
def __init__(self, app, max_requests: int, window_seconds: int):
super().__init__(app)
self.max_requests = max_requests
self.window_seconds = window_seconds

async def dispatch(self, request: Request, call_next):
client_id = request.client.host

if not rate_limiter.fixed_window_limit(client_id, self.max_requests, self.window_seconds):
raise HTTPException(
status_code=429,
detail={
'error': 'Rate limit exceeded',
'retry_after': self.window_seconds
}
)

response = await call_next(request)
return response

app.add_middleware(RateLimitMiddleware, max_requests=100, window_seconds=60)

@app.get('/api/data')
async def get_data():
return {'data': 'Hello, World!'}

五、多级限流策略

5.1 多层限流架构

class MultiLevelRateLimiter:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client
self.limiters = {
'ip': RedisRateLimiter(redis_client),
'user': RedisRateLimiter(redis_client),
'global': RedisRateLimiter(redis_client)
}

def check_all_limits(self, ip: str, user_id: str = None) -> bool:
checks = [
self.limiters['ip'].fixed_window_limit(f"ip:{ip}", 100, 60),
self.limiters['global'].fixed_window_limit("global", 1000, 60)
]

if user_id:
checks.append(
self.limiters['user'].fixed_window_limit(f"user:{user_id}", 50, 60)
)

return all(checks)

5.2 动态限流策略

class DynamicRateLimiter:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client
self.base_limit = 100
self.min_limit = 10
self.max_limit = 1000

def get_dynamic_limit(self) -> int:
cpu_usage = self._get_cpu_usage()
memory_usage = self._get_memory_usage()

load_factor = 1.0

if cpu_usage > 80:
load_factor *= 0.5
elif cpu_usage > 60:
load_factor *= 0.8

if memory_usage > 80:
load_factor *= 0.5
elif memory_usage > 60:
load_factor *= 0.8

return max(self.min_limit, min(self.max_limit, int(self.base_limit * load_factor)))

def is_allowed(self, key: str) -> bool:
limit = self.get_dynamic_limit()
return self.redis_client.incr(key) <= limit

def _get_cpu_usage(self) -> float:
return 40.0

def _get_memory_usage(self) -> float:
return 50.0

六、实际业务场景应用

6.1 API网关限流

class APIGatewayRateLimiter:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client
self.rules = {
'/api/v1/users': {'max_requests': 100, 'window_seconds': 60},
'/api/v1/orders': {'max_requests': 50, 'window_seconds': 60},
'/api/v1/payments': {'max_requests': 20, 'window_seconds': 60}
}

def check_rate_limit(self, endpoint: str, client_id: str) -> bool:
if endpoint not in self.rules:
return True

rule = self.rules[endpoint]
key = f"rate_limit:{endpoint}:{client_id}"

return self.redis_client.incr(key) <= rule['max_requests']

def set_rule(self, endpoint: str, max_requests: int, window_seconds: int):
self.rules[endpoint] = {
'max_requests': max_requests,
'window_seconds': window_seconds
}

6.2 用户级限流

class UserRateLimiter:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client

def check_user_limit(self, user_id: str, limit_type: str) -> bool:
limits = {
'daily': {'max_requests': 1000, 'window_seconds': 86400},
'hourly': {'max_requests': 100, 'window_seconds': 3600},
'minute': {'max_requests': 10, 'window_seconds': 60}
}

if limit_type not in limits:
return True

rule = limits[limit_type]
key = f"user_limit:{user_id}:{limit_type}"

return self.redis_client.incr(key) <= rule['max_requests']

七、限流监控与告警

7.1 限流监控系统

class RateLimitMonitor:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client

def get_rate_limit_stats(self, pattern: str = '*') -> dict:
stats = {}

for key in self.redis_client.keys(f"rate_limit:{pattern}"):
key_str = key.decode('utf-8')
count = int(self.redis_client.get(key) or 0)
ttl = self.redis_client.ttl(key)

stats[key_str] = {
'count': count,
'ttl': ttl
}

return stats

def get_top_limiters(self, limit: int = 10) -> list:
stats = self.get_rate_limit_stats()
sorted_stats = sorted(stats.items(), key=lambda x: x[1]['count'], reverse=True)

return sorted_stats[:limit]

def check_anomalies(self, threshold: int = 90) -> list:
anomalies = []
stats = self.get_rate_limit_stats()

for key, info in stats.items():
percentage = (info['count'] / self._get_limit(key)) * 100
if percentage > threshold:
anomalies.append({
'key': key,
'percentage': percentage,
'count': info['count']
})

return anomalies

7.2 限流告警系统

class RateLimitAlertSystem:
def __init__(self, monitor: RateLimitMonitor):
self.monitor = monitor
self.alert_threshold = 90
self.alert_history = []

def check_and_alert(self):
anomalies = self.monitor.check_anomalies(self.alert_threshold)

for anomaly in anomalies:
if anomaly['key'] not in self.alert_history:
self._send_alert(anomaly)
self.alert_history.append(anomaly['key'])

self._clean_alert_history()

def _send_alert(self, anomaly: dict):
print(f"ALERT: Rate limit approaching for {anomaly['key']} "
f"- {anomaly['percentage']:.2f}% utilized")

def _clean_alert_history(self):
if len(self.alert_history) > 100:
self.alert_history = self.alert_history[-100:]

八、性能优化策略

8.1 本地缓存优化

class CachedRateLimiter:
def __init__(self, redis_client: redis.Redis, local_cache_size: int = 1000):
self.redis_client = redis_client
self.local_cache = {}
self.local_cache_size = local_cache_size

def is_allowed(self, key: str, max_requests: int, window_seconds: int) -> bool:
if key in self.local_cache:
entry = self.local_cache[key]
if entry['window'] == int(time.time() // window_seconds):
if entry['count'] < max_requests:
entry['count'] += 1
return True
return False

result = self._redis_check(key, max_requests, window_seconds)

if result:
if len(self.local_cache) >= self.local_cache_size:
oldest_key = next(iter(self.local_cache))
del self.local_cache[oldest_key]

self.local_cache[key] = {
'count': 1,
'window': int(time.time() // window_seconds)
}

return result

def _redis_check(self, key: str, max_requests: int, window_seconds: int) -> bool:
window_key = f"rate_limit:{key}:{int(time.time() // window_seconds)}"

count = self.redis_client.incr(window_key)
if count == 1:
self.redis_client.expire(window_key, window_seconds)

return count <= max_requests

8.2 批量操作优化

class BatchRateLimiter:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client

def check_batch_limits(self, keys: list, max_requests: int, window_seconds: int) -> dict:
now = time.time()
window = int(now // window_seconds)

pipeline = self.redis_client.pipeline()
window_keys = [f"rate_limit:{key}:{window}" for key in keys]

for window_key in window_keys:
pipeline.incr(window_key)
pipeline.expire(window_key, window_seconds)

results = pipeline.execute()
counts = results[::2]

return {key: count <= max_requests for key, count in zip(keys, counts)}

总结

API限流是构建高可用系统的关键技术。通过本文的学习，你应该掌握了以下核心要点：

作为从Python转向Rust的后端开发者，掌握限流策略对于构建健壮的API服务至关重要。后续文章将探讨如何在Rust中实现高性能限流系统。