Python脚本限流算法有哪些实现方式

本文目录导读：

1. 令牌桶算法（Token Bucket）
2. 漏桶算法（Leaky Bucket）
3. 固定窗口计数器（Fixed Window Counter）
4. 滑动窗口日志（Sliding Window Log）
5. 滑动窗口计数器（Sliding Window Counter）
6. 使用第三方库（推荐生产环境）
7. 选择建议

Python 中实现限流算法的常见方式有以下几种,按复杂度从低到高排列：

令牌桶算法（Token Bucket）

最常用的限流算法,允许一定的突发流量。

import time
import threading
class TokenBucket:
    def __init__(self, rate, capacity):
        self.rate = rate  # 令牌生成速率（个/秒）
        self.capacity = capacity  # 桶容量
        self.tokens = capacity  # 当前令牌数
        self.last_time = time.time()
        self.lock = threading.Lock()
    def consume(self, tokens=1):
        with self.lock:
            now = time.time()
            # 计算这段时间生成的令牌
            elapsed = now - self.last_time
            self.tokens = min(self.capacity, 
                             self.tokens + elapsed * self.rate)
            self.last_time = now
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

漏桶算法（Leaky Bucket）

请求以固定速率被处理,适合平滑流量。

import time
import threading
from collections import deque
class LeakyBucket:
    def __init__(self, rate, capacity):
        self.rate = rate  # 处理速率（请求/秒）
        self.capacity = capacity  # 桶容量
        self.water = 0
        self.last_time = time.time()
        self.lock = threading.Lock()
    def allow_request(self):
        with self.lock:
            now = time.time()
            # 漏水
            elapsed = now - self.last_time
            self.water = max(0, self.water - elapsed * self.rate)
            self.last_time = now
            if self.water < self.capacity:
                self.water += 1
                return True
            return False

固定窗口计数器（Fixed Window Counter）

最简单的实现，适合精确控制每分钟/每小时的请求数。

import time
import threading
class FixedWindowCounter:
    def __init__(self, limit, window_size):
        self.limit = limit  # 窗口内最大请求数
        self.window_size = window_size  # 窗口大小（秒）
        self.window_start = 0
        self.counter = 0
        self.lock = threading.Lock()
    def allow_request(self):
        with self.lock:
            now = time.time()
            # 检查是否进入新窗口
            if now - self.window_start >= self.window_size:
                self.window_start = now
                self.counter = 0
            if self.counter < self.limit:
                self.counter += 1
                return True
            return False

滑动窗口日志（Sliding Window Log）

精确记录每个请求的时间戳,更精确但内存消耗大。

import time
import threading
from collections import deque
class SlidingWindowLog:
    def __init__(self, limit, window_size):
        self.limit = limit  # 窗口内最大请求数
        self.window_size = window_size  # 窗口大小（秒）
        self.log = deque()
        self.lock = threading.Lock()
    def allow_request(self):
        with self.lock:
            now = time.time()
            window_start = now - self.window_size
            # 移除过期记录
            while self.log and self.log[0] < window_start:
                self.log.popleft()
            if len(self.log) < self.limit:
                self.log.append(now)
                return True
            return False

滑动窗口计数器（Sliding Window Counter）

折中方案，比固定窗口精确,比滑动日志节省内存。

import time
import threading
class SlidingWindowCounter:
    def __init__(self, limit, window_size):
        self.limit = limit  # 窗口内最大请求数
        self.window_size = window_size  # 窗口大小（秒）
        self.prev_counter = 0
        self.curr_counter = 0
        self.prev_window_start = 0
        self.curr_window_start = time.time()
        self.lock = threading.Lock()
    def allow_request(self):
        with self.lock:
            now = time.time()
            elapsed = now - self.curr_window_start
            if elapsed >= self.window_size:
                # 进入新窗口
                self.prev_window_start = self.curr_window_start
                self.curr_window_start = now
                self.prev_counter = self.curr_counter
                self.curr_counter = 0
                elapsed = 0
            # 计算当前滑动窗口的估计值
            weight = elapsed / self.window_size
            estimated_count = self.prev_counter * (1 - weight) + self.curr_counter
            if estimated_count < self.limit:
                self.curr_counter += 1
                return True
            return False

使用第三方库（推荐生产环境）

aio-limiter（异步）

from aiolimiter import AsyncLimiter
limiter = AsyncLimiter(100, 10)  # 每秒100个请求
async with limiter:
    # 执行需要限流的操作
    pass

limits（同步/异步）

from limits import strategies, storage, RateLimitItemPerSecond
# 内存存储
mem_storage = storage.MemoryStorage()
strategy = strategies.MovingWindowRateLimiter(mem_storage)
# 定义限流规则：每秒10个请求
rate_limit = RateLimitItemPerSecond(10, 1)
if strategy.hit(rate_limit, "user_123"):
    # 允许请求
    pass

redis-rate-limiter（分布式）

import redis
from redis_rate_limiter import RateLimiter
redis_client = redis.Redis()
rate_limiter = RateLimiter(
    redis_client,
    max_requests=100,
    window=60  # 60秒窗口
)
if rate_limiter.acquire("api_key_123"):
    # 允许请求
    pass

选择建议

面试或学习时建议掌握令牌桶和滑动窗口,它们是最实用和问得最多的限流算法。