Python限流常用固定窗口、滑动窗口、令牌桶三种算法：固定窗口简单高效但有临界突发问题；滑动窗口精度高、内存开销略大；令牌桶支持突发流量，推荐生产使用，分布式场景建议Redis实现。
Python 中实现限流（Rate Limiting）常用算法有三种：固定窗口、滑动窗口、令牌桶。选择哪种取决于你对精度、内存占用和并发安全的要求。
固定窗口计数器（简单高效，适合粗粒度限流）
将时间划分为固定长度的窗口（如 1 分钟），每个窗口独立计数。实现简单，但存在临界突发问题（窗口切换时可能翻倍放行）。
- 用字典或 Redis 存储 key → {窗口开始时间, 请求次数}，每次请求检查是否在当前窗口内
- Python 示例（内存版，非线程安全，生产建议用 Redis 或加锁）：
from time import time
from collections import defaultdict
<p>class FixedWindowLimiter:
def <strong>init</strong>(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = defaultdict(list) # key → [timestamp1, timestamp2, ...]</p><pre class="brush:php;toolbar:false;">def is_allowed(self, key: str) -> bool:
now = int(time())
window_start = (now // self.window_seconds) * self.window_seconds
# 只保留当前窗口内的请求时间戳
self.requests[key] = [
t for t in self.requests[key]
if window_start <= t < window_start + self.window_seconds
]
if len(self.requests[key]) < self.max_requests:
self.requests[key].append(now)
return True
return False使用示例
limiter = FixedWindowLimiter(max_requests=5, window_seconds=60) print(limiter.is_allowed("user:123")) # True
滑动窗口日志（精度高，内存开销略大）
记录每次请求的精确时间戳，判断最近 N 秒内请求数是否超限。比固定窗口更平滑，但需维护有序时间序列（可用 deque 优化）。
- 使用
collections.deque保存时间戳，自动左弹出过期项 - 每次请求前清理过期时间戳，再判断长度
from collections import deque
from time import time
<p>class SlidingWindowLimiter:
def <strong>init</strong>(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = defaultdict(lambda: deque())</p><pre class="brush:php;toolbar:false;">def is_allowed(self, key: str) -> bool:
now = time()
window_start = now - self.window_seconds
# 清理过期请求
while self.requests[key] and self.requests[key][0] < window_start:
self.requests[key].popleft()
if len(self.requests[key]) < self.max_requests:
self.requests[key].append(now)
return True
return False
立即学习“Python免费学习笔记（深入）”；
令牌桶（支持突发流量，推荐生产使用）
以恒定速率向桶中添加令牌，每次请求消耗一个令牌。桶有容量上限，允许短时突发（只要桶未空）。
- 关键参数：桶容量（burst）、填充速率（rate per second）
- 每次请求计算自上次以来应新增的令牌数，更新剩余令牌并判断
- 注意浮点精度和并发安全（多线程建议用 threading.Lock 或用 Redis 实现）
from time import time
from threading import Lock
<p>class TokenBucketLimiter:
def <strong>init</strong>(self, capacity: int, fill_rate: float):
self.capacity = capacity
self.fill_rate = fill_rate # tokens per second
self.tokens = capacity
self.last_refill = time()
self.lock = Lock()</p><pre class="brush:php;toolbar:false;">def _refill(self):
now = time()
elapsed = now - self.last_refill
new_tokens = elapsed * self.fill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
def is_allowed(self) -> bool:
with self.lock:
self._refill()
if self.tokens >= 1:
self.tokens -= 1
return True
return False使用示例：每秒最多 2 个请求，最多可突发 5 个
limiter = TokenBucketLimiter(capacity=5, fill_rate=2.0)
生产环境建议
单机服务可用上述内存实现（注意加锁）；分布式场景必须依赖共享存储，推荐 Redis：
- 固定窗口：用
INCR+EXPIRE实现原子计数 - 滑动窗口：用
ZSET存时间戳，ZREMRANGEBYSCORE清理过期项 - 令牌桶：Redis + Lua 脚本保证原子性（如官方推荐的 rate limiting with Redis）
不复杂但容易忽略：务必设置合理的 key 粒度（如 user_id、ip、api_path 组合），并考虑冷启动（首次请求时桶是否预热）和时钟漂移问题。
