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 组合），并考虑冷启动（首次请求时桶是否预热）和时钟漂移问题。