当前位置：首页 > 文章列表 > 文章 > python教程 > Python 实现限流算法示例

Python 实现限流算法示例

2026-05-03 18:31:31 0浏览收藏

在文章实战开发的过程中，我们经常会遇到一些这样那样的问题，然后要卡好半天，等问题解决了才发现原来一些细节知识点还是没有掌握好。今天golang学习网就整理分享《Python 实现限流算法示例》，聊聊，希望可以帮助到正在努力赚钱的你。

Python限流常用固定窗口、滑动窗口、令牌桶三种算法：固定窗口简单高效但有临界突发问题；滑动窗口精度高、内存开销略大；令牌桶支持突发流量，推荐生产使用，分布式场景建议Redis实现。

Python 实现限流算法示例

Python 中实现限流（Rate Limiting）常用算法有三种：固定窗口、滑动窗口、令牌桶。选择哪种取决于你对精度、内存占用和并发安全的要求。

固定窗口计数器（简单高效，适合粗粒度限流）

将时间划分为固定长度的窗口（如 1 分钟），每个窗口独立计数。实现简单，但存在临界突发问题（窗口切换时可能翻倍放行）。

用字典或 Redis 存储 key → {窗口开始时间, 请求次数}，每次请求检查是否在当前窗口内
Python 示例（内存版，非线程安全，生产建议用 Redis 或加锁）：

from time import time
from collections import defaultdict
class FixedWindowLimiter:
def init(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = defaultdict(list)  # key → [timestamp1, timestamp2, ...]
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
class SlidingWindowLimiter:
def init(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = defaultdict(lambda: deque())
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

令牌桶（支持突发流量，推荐生产使用）

以恒定速率向桶中添加令牌，每次请求消耗一个令牌。桶有容量上限，允许短时突发（只要桶未空）。

关键参数：桶容量（burst）、填充速率（rate per second）
每次请求计算自上次以来应新增的令牌数，更新剩余令牌并判断
注意浮点精度和并发安全（多线程建议用 threading.Lock 或用 Redis 实现）

from time import time
from threading import Lock
class TokenBucketLimiter:
def init(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()
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)