API 限流实战：从被刷爆到稳如磐石
上个月，我们的 API 被恶意刷了 200 万次请求，服务器直接宕机。
经过一周的优化，现在即使面对 10 倍流量也能稳定运行。
问题背景
被攻击的那天
CODE
2026-06-15 03:22:15 [ERROR] Connection pool exhausted
2026-06-15 03:22:16 [ERROR] Database timeout
2026-06-15 03:22:17 [ERROR] 503 Service Unavailable监控显示：
- 正常流量：5000 请求/分钟
- 攻击流量：50000 请求/分钟
- 服务器 CPU：100%
- 数据库连接：全部占满
根本原因
1. 没有限流：任何人都可以无限请求
2. 没有缓存：每次请求都查数据库
3. 没有降级：压力大了直接崩溃
限流策略
策略 1：固定窗口限流
最简单的限流方式。
PYTHON
from datetime import datetime
import redis
class FixedWindowRateLimiter:
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def is_allowed(self, key: str, limit: int, window: int) -> bool:
"""
key: 限流键（如 user_id 或 ip）
limit: 窗口内最大请求数
window: 窗口大小（秒）
"""
current_minute = datetime.now().strftime("%Y%m%d%H%M")
redis_key = f"rate:{key}:{current_minute}"
count = self.redis.incr(redis_key)
if count == 1:
self.redis.expire(redis_key, window)
return count <= limit
# 使用
limiter = FixedWindowRateLimiter(redis_client)
@app.route("/api/data")
def get_data():
user_id = get_current_user_id()
if not limiter.is_allowed(user_id, limit=100, window=60):
return jsonify({"error": "Too many requests"}), 429
return jsonify({"data": "..."})问题：窗口边界可能突发 2 倍流量。
策略 2：滑动窗口限流
解决固定窗口的边界问题。
PYTHON
class SlidingWindowRateLimiter:
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def is_allowed(self, key: str, limit: int, window: int) -> bool:
now = time.time()
window_start = now - window
redis_key = f"rate:{key}"
# 使用 sorted set 记录请求时间
pipe = self.redis.pipeline()
pipe.zremrangebyscore(redis_key, 0, window_start)
pipe.zadd(redis_key, {str(now): now})
pipe.zcard(redis_key)
pipe.expire(redis_key, window)
results = pipe.execute()
count = results[2]
return count <= limit
# 使用
limiter = SlidingWindowRateLimiter(redis_client)
@app.route("/api/data")
def get_data():
user_id = get_current_user_id()
if not limiter.is_allowed(user_id, limit=100, window=60):
return jsonify({"error": "Too many requests"}), 429
return jsonify({"data": "..."})策略 3：令牌桶算法
允许一定程度的突发流量。
PYTHON
import time
class TokenBucketRateLimiter:
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def is_allowed(self, key: str, capacity: int, refill_rate: float) -> bool:
"""
capacity: 桶容量
refill_rate: 每秒补充的令牌数
"""
redis_key = f"token:{key}"
# 获取当前状态
data = self.redis.hgetall(redis_key)
now = time.time()
if not data:
# 首次请求，满桶
tokens = capacity - 1
last_refill = now
else:
tokens = float(data[b'tokens'])
last_refill = float(data[b'last_refill'])
# 计算补充的令牌
elapsed = now - last_refill
tokens = min(capacity, tokens + elapsed * refill_rate)
if tokens >= 1:
tokens -= 1
else:
return False
# 保存状态
self.redis.hset(redis_key, mapping={
'tokens': tokens,
'last_refill': now
})
self.redis.expire(redis_key, 60)
return True
# 使用
limiter = TokenBucketRateLimiter(redis_client)
@app.route("/api/data")
def get_data():
user_id = get_current_user_id()
# 允许 100 个突发，每秒补充 10 个
if not limiter.is_allowed(user_id, capacity=100, refill_rate=10):
return jsonify({"error": "Too many requests"}), 429
return jsonify({"data": "..."})多层限流架构
层级设计
CODE
┌─────────────────────────────────────┐
│ Layer 1: Nginx 限流 │
│ - IP 级别 │
│ - 1000 req/min │
├─────────────────────────────────────┤
│ Layer 2: 应用层限流 │
│ - 用户级别 │
│ - 100 req/min │
├─────────────────────────────────────┤
│ Layer 3: 接口级别限流 │
│ - 特定接口 │
│ - 10 req/min │
└─────────────────────────────────────┘Nginx 配置
NGINX
# /etc/nginx/nginx.conf
# 定义限流区域
limit_req_zone $binary_remote_addr zone=ip_limit:10m rate=100r/s;
limit_req_zone $http_x_user_id zone=user_limit:10m rate=50r/s;
server {
listen 80;
# IP 级别限流
location /api/ {
limit_req zone=ip_limit burst=50 nodelay;
limit_req_status 429;
proxy_pass http://backend;
}
# 用户级别限流
location /api/user/ {
limit_req zone=user_limit burst=20 nodelay;
limit_req_status 429;
proxy_pass http://backend;
}
# 特定接口限流
location /api/expensive/ {
limit_req zone=ip_limit rate=5r/s burst=10 nodelay;
limit_req_status 429;
proxy_pass http://backend;
}
}应用层限流
PYTHON
from functools import wraps
from flask import request, jsonify
def rate_limit(limit: int, window: int, key_func=None):
"""限流装饰器"""
def decorator(f):
@wraps(f)
def wrapped(*args, **kwargs):
# 获取限流键
if key_func:
key = key_func()
else:
key = request.headers.get('X-User-ID') or request.remote_addr
# 检查限流
if not limiter.is_allowed(key, limit, window):
return jsonify({
"error": "Too many requests",
"retry_after": window
}), 429
return f(*args, **kwargs)
return wrapped
return decorator
# 使用
@app.route("/api/data")
@rate_limit(limit=100, window=60)
def get_data():
return jsonify({"data": "..."})
@app.route("/api/expensive")
@rate_limit(limit=10, window=60, key_func=lambda: f"expensive:{get_current_user_id()}")
def expensive_operation():
return jsonify({"result": "..."})降级策略
策略 1：缓存降级
PYTHON
from functools import lru_cache
import time
class CacheWithFallback:
def __init__(self, redis_client):
self.redis = redis_client
self.local_cache = {}
def get(self, key, fetch_func, ttl=300):
# 1. 尝试 Redis 缓存
try:
cached = self.redis.get(key)
if cached:
return json.loads(cached)
except redis.RedisError:
pass # Redis 不可用，继续
# 2. 尝试本地缓存
if key in self.local_cache:
data, timestamp = self.local_cache[key]
if time.time() - timestamp < ttl:
return data
# 3. 获取新数据
try:
data = fetch_func()
# 写入缓存
try:
self.redis.setex(key, ttl, json.dumps(data))
except redis.RedisError:
pass
self.local_cache[key] = (data, time.time())
return data
except Exception:
# 4. 返回过期缓存
if key in self.local_cache:
return self.local_cache[key][0]
raise策略 2：功能降级
PYTHON
class FeatureFlags:
def __init__(self):
self.degraded_features = set()
def degrade(self, feature: str):
self.degraded_features.add(feature)
def restore(self, feature: str):
self.degraded_features.discard(feature)
def is_degraded(self, feature: str) -> bool:
return feature in self.degraded_features
flags = FeatureFlags()
@app.route("/api/recommendations")
def get_recommendations():
if flags.is_degraded("recommendations"):
# 返回简单推荐
return jsonify({"items": get_popular_items()})
# 正常推荐逻辑
return jsonify({"items": get_personalized_recommendations()})
# 监控系统自动降级
def check_system_health():
cpu = get_cpu_usage()
if cpu > 90:
flags.degrade("recommendations")
flags.degrade("analytics")
elif cpu < 70:
flags.restore("recommendations")
flags.restore("analytics")监控与告警
关键指标
PYTHON
from prometheus_client import Counter, Histogram
# 限流触发次数
rate_limit_hits = Counter(
'rate_limit_hits_total',
'Number of rate limit hits',
['endpoint', 'limit_type']
)
# 请求延迟
request_latency = Histogram(
'request_latency_seconds',
'Request latency',
['endpoint']
)
# 在限流装饰器中记录
def rate_limit(limit: int, window: int):
def decorator(f):
@wraps(f)
def wrapped(*args, **kwargs):
key = get_rate_limit_key()
if not limiter.is_allowed(key, limit, window):
rate_limit_hits.labels(
endpoint=request.path,
limit_type='user'
).inc()
return jsonify({"error": "Too many requests"}), 429
return f(*args, **kwargs)
return wrapped
return decorator告警规则
YAML
# Prometheus alerting rules
groups:
- name: rate_limit_alerts
rules:
- alert: HighRateLimitHits
expr: rate(rate_limit_hits_total[5m]) > 100
for: 5m
labels:
severity: warning
annotations:
summary: "High rate limit hit rate"
description: "Rate limit hits > 100/s for 5 minutes"
- alert: PossibleDDoS
expr: rate(requests_total[1m]) > 10000
for: 2m
labels:
severity: critical
annotations:
summary: "Possible DDoS attack"
description: "Request rate > 10000/s"效果对比
| 指标 | 优化前 | 优化后 |
|------|--------|--------|
| 最大 QPS | 5000（崩溃） | 50000（稳定） |
| 限流精度 | 无 | 99.9% |
| 降级响应 | 无 | < 100ms |
| 恢复时间 | 30 分钟 | < 1 分钟 |
总结
限流的核心原则：
1. 多层防护：Nginx + 应用层 + 接口层
2. 灵活策略：固定窗口、滑动窗口、令牌桶按需选择
3. 优雅降级：压力大了自动降级非核心功能
4. 实时监控：发现问题立即告警
做好这些，API 就能稳如磐石。
优化时间：2026年6月
攻击规模：200 万请求/分钟
优化效果：10 倍流量稳定运行
#API #限流 #RateLimit #高可用
读者评论 5