本文目录导读：
限制访问频率（即限流）通常是为了防止恶意攻击、爬虫滥用或保证服务器稳定运行，实现方式根据场景（服务端/客户端、单机/分布式、全局/用户维度）有所不同。
以下是几种常用的实现脚本（Python示例，但逻辑通用）,以及相应的策略：
基于时间窗口的简单计数
这是最直接的思路，用一个计数器记录单位时间内的访问次数,超限则拒绝。
适用场景： 单机、轻量级、对精度要求不高的场景。
伪代码逻辑：
import time
class SimpleRateLimiter:
def __init__(self, max_requests=10, window_seconds=60):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = [] # 存储每个请求的时间戳
def is_allowed(self):
now = time.time()
# 1. 移除窗口之外的旧记录
self.requests = [t for t in self.requests if now - t < self.window_seconds]
# 2. 检查当前窗口内的请求数
if len(self.requests) < self.max_requests:
self.requests.append(now)
return True
else:
return False
# 使用示例
limiter = SimpleRateLimiter(max_requests=5, window_seconds=10)
# 模拟请求
if limiter.is_allowed():
print("正常处理")
else:
print("请求过于频繁，请稍后再试")
缺点： 在窗口边界可能出现突刺（例如第一秒请求完成，下一个窗口刚开始瞬间请求大量涌入）。
基于滑动日志（Sliding Log）
解决窗口边界突刺问题，记录每次请求的时间戳,并计算过去指定时间内的总量。
适用场景： 需要更精确的限流的本地场景。
逻辑： 使用一个有序集合（Sorted Set）或者列表记录所有时间戳，每次请求时，删除掉比当前时间早一个窗口的记录，检查剩余数量，你可以用 Redis 的 ZSET（有序集合）实现分布式版本。
Redis 实现思路（推荐用于分布式）：
ZADD key timestamp member(member 通常设为唯一 ID 或时间戳本身)ZREMRANGEBYSCORE key -inf (now - window)ZCARD key获取当前集合大小- 如果小于阈值，批准；否则拒绝。
令牌桶算法（Token Bucket，最推荐）
原理： 一个固定容量的桶按固定速率放入令牌，请求需要先消耗一个令牌才能被处理，如果桶空了,请求被拒绝。
优点： 允许有突发流量（桶里可以积累令牌），同时又限制了平均流量（注入速率固定）。
Python 单机实现：
import time
class TokenBucket:
def __init__(self, rate=1, capacity=10):
self.rate = rate # 令牌放入速率（个/秒）
self.capacity = capacity # 桶容量
self.tokens = capacity # 当前令牌数
self.last_refill_time = time.time()
def is_allowed(self):
now = time.time()
# 1. 计算这段时间应该产生的令牌数
elapsed = now - self.last_refill_time
new_tokens = elapsed * self.rate
# 2. 补充令牌（不能超过容量）
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill_time = now
# 3. 消耗令牌
if self.tokens >= 1:
self.tokens -= 1
return True
return False
# 示例：每秒允许2个请求，最多积累5个突发令牌
limiter = TokenBucket(rate=2, capacity=5)
分布式实现： 可以使用 Redis 的 INCR + 过期时间，或者使用 Lua 脚本封装原子操作，更成熟的方案是集成如 Guava RateLimiter（Java）或 Limiter 中间件。
IP/用户维度限流（最重要的安全措施）
通常你需要区分不同来源（IP 或用户 ID）,而不是全局限制。
常见策略：
- 单 IP 限流： 限制每个 IP 每分钟 60 次。
- 用户维度： 限制每个已登录用户每分钟 100 次。
实现方式：
import time
from collections import defaultdict
class IPBasedLimiter:
def __init__(self, max_requests=100, window_seconds=60):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.ip_records = defaultdict(list) # {ip: [timestamp1, timestamp2]}
def is_allowed(self, ip):
now = time.time()
# 清理该 IP 下的过期记录
self.ip_records[ip] = [t for t in self.ip_records[ip] if now - t < self.window_seconds]
if len(self.ip_records[ip]) < self.max_requests:
self.ip_records[ip].append(now)
return True
return False
# 使用示例
limiter = IPBasedLimiter(max_requests=5, window_seconds=10)
def handle_request(request):
client_ip = request.get('remote_ip', 'unknown')
if limiter.is_allowed(client_ip):
print(f"处理来自 {client_ip} 的请求")
else:
print(f"拒绝来自 {client_ip} 的请求 (频率过高)")
注意： 如果用户量巨大且在分布式环境下，IP 记录不能存在本地字典，必须存在 Redis 等共享存储里。
实战场景下的分层架构建议
在实际的 WEB 服务或 API 脚本中，建议从近到远设置三层防护：
-
网关层（最前沿）：
-
使用 Nginx 的
limit_req_zone模块或 Cloudflare 的 Rate Limiting。 -
优点： 性能高,不消耗应用服务器资源。
-
Nginx 配置示例：
limit_req_zone $binary_remote_addr zone=one:10m rate=10r/s; server { location /api/ { limit_req zone=one burst=20 nodelay; # 允许突发 20 个请求 } }
-
-
应用层（业务逻辑）：
- 使用 Redis + Lua 脚本实现令牌桶或滑动窗口。
- 用于处理更复杂的策略（例如按用户 ID 限流、按 API 路径限流）。
-
客户端（前端兜底？）：
不做安全上的信任，只用于提升用户体验（如按钮 1 秒内不可重复点击）。
重要提示
- 拒绝响应码： 当限制频率时，HTTP 最好返回 429 Too Many Requests。
- 返回头部信息： 建议在响应头中包含
X-RateLimit-Limit（总限额）、X-RateLimit-Remaining（剩余次数）、Retry-After（需要等待多少秒后重试）,方便客户端做客户端流控。 - 滑动窗口 vs 固定窗口： 除了 Redis ZSET 实现滑动窗口，也可以用漏桶算法（Leaky Bucket，强制平滑速率）或令牌桶（允许突发）。
- 分布式锁： 如果使用 Redis 计数器，在并发极高时需要考虑原子操作（可以使用
INCR+EXPIRE，但不同窗口切换时可能有竞态问题，建议使用 Lua 或 Redlock 确保一致性）。
总结建议
- 简单测试脚本： 使用本地
TokenBucket类。 - 生产环境单机应用： 使用 Nginx 限流 + 框架自带的限流中间件（Flask-Limiter 支持 Redis 后端）。
- 分布式高并发系统： 使用 Redis 滑动窗口（ZSET）或令牌桶（配合 Lua），并结合 Nginx 网关层限流。
