深入理解限流算法：从令牌桶到滑动窗口的工程实践

在高并发系统中，限流（Rate Limiting）是保护服务稳定性的核心手段之一。无论你是做 API 网关、微服务还是公共接口，都需要一套靠谱的限流方案。今天我们从原理到实现，聊聊几种主流限流��法。

为什么需要限流？

• 防止服务雪崩：突发流量打满资源，一个服务挂了拖垮整条链路
• 保障公平性：避免单个用户/IP 占用过多资源
• API 计费：SaaS 产品常见的按量计费模型依赖精确限流
• 安全防护：防止暴力破解、爬虫滥用等

四种经典算法

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

最简单的方案：把时间分成固定窗口（比如每秒），在窗口内计数。

import time

class FixedWindowLimiter:
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.window_start = time.time()
        self.count = 0
    
    def allow(self):
        now = time.time()
        if now - self.window_start >= self.window_seconds:
            self.window_start = now
            self.count = 0
        if self.count < self.max_requests:
            self.count += 1
            return True
        return False

问题：窗口边界处可能出现 2 倍突发。比如限制每秒 100 次，在 0.9s 发了 100 次，1.0s 又发 100 次——实际 0.2 秒内通过了 200 次请求。

2. 滑动窗口日志（Sliding Window Log）

记录每个请求的精确时间戳，每次判断时清理过期记录：

from collections import deque
import time

class SlidingWindowLog:
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.log = deque()
    
    def allow(self):
        now = time.time()
        # 清理过期记录
        while self.log and self.log[0] <= now - self.window_seconds:
            self.log.popleft()
        if len(self.log) < self.max_requests:
            self.log.append(now)
            return True
        return False

优点：精确，没有边界问题。缺点：内存开销大，每个请求都要存时间戳。

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

折中方案——用两个相邻窗口的加权计数来近似滑动窗口：

当前窗口权重 = 当前窗口已过时间 / 窗口大小
估算请求数 = 上一窗口计数 × (1 - 当前权重) + 当前窗口计数

内存只需要两个计数器，精度接近滑动日志。Redis 的很多限流方案用的就是这个思路。

4. 令牌桶（Token Bucket）

最灵活也最常用的算法：

import time

class TokenBucket:
    def __init__(self, rate, capacity):
        self.rate = rate          # 每秒生成令牌数
        self.capacity = capacity  # 桶最大容量
        self.tokens = capacity
        self.last_refill = time.time()
    
    def allow(self, tokens=1):
        now = time.time()
        # 补充令牌
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_refill = now
        
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

核心优势：

• 允许一定程度的突发流量（桶里攒的令牌可以一次性用掉）
• 长期来看严格控制平均速率
• capacity 控制突发上限，rate 控制稳态速率

分布式场景怎么做？

单机限流用内存就行，但微服务架构下需要全局限流。常见方案：

Redis + Lua 脚本

-- 滑动窗口限流 Lua 脚本
local key = KEYS[1]
local window = tonumber(ARGV[1])
local limit = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

redis.call("ZREMRANGEBYSCORE", key, 0, now - window)
local count = redis.call("ZCARD", key)

if count < limit then
    redis.call("ZADD", key, now, now .. math.random())
    redis.call("EXPIRE", key, window)
    return 1
end
return 0

用 Sorted Set 存时间戳，Lua 保证原子性。简单高效，生产环境广泛使用。

算法选型建议

场景	推荐算法	原因
API 网关全局限流	令牌桶	允许突发，平滑控制
用户级配额	滑动窗口计数器	精度够，内存省
登录防暴力破解	固定窗口	实现简单，够用
精确计费	滑动窗口日志	最精确

实际部署注意事项

1. 限流粒度：按 IP？按用户？按 API Key？按接口？组合使用最靠谱
2. 返回友好信息：HTTP 429 + Retry-After 头，别让客户端瞎猜
3. 监控告警：限流触发率是重要指标，太高说明容量不够或有异常流量
4. 降级策略：限流不是万能的，配合熔断、降级一起用

限流看起来简单，但细节很多。建议从令牌桶开始，覆盖 80% 的场景，再根据具体需求引入其他方案。

---

写代码不难，写出稳定的代码才难。限流是系统韧性的第一道防线，值得认真对待。