rate-limiting-algorithms

Rate Limiting Algorithms Skill — 限流算法

何时使用

• 公开 API 防滥用 / 防爬虫 / 防 DDoS
• 第三方服务调用（自家被动限流避免成本爆炸）
• 设计 API quota / billing 体系
• 选择 Nginx / Redis / 应用层限流的组合
• 排查"客户感觉被误伤"

一、六种核心算法

1. Fixed Window（固定窗口）

[ 0:00 - 0:59 ] 计数 100 次
[ 1:00 - 1:59 ] 计数清零，重新计 100 次

# Redis 实现
key = f"ratelimit:{user_id}:{minute}"
count = redis.incr(key)
if count == 1: redis.expire(key, 60)
if count > 100: reject()

优点：简单 / 内存少 缺点：边界突发：0:59 用 100 次 + 1:00 立刻又 100 次 = 1 秒内 200 次，可能压垮下游

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

# 每次请求记 timestamp 到 sorted set
redis.zadd(key, {req_id: now})
redis.zremrangebyscore(key, 0, now - 60)   # 清 60s 前
count = redis.zcard(key)
if count > 100: reject()

优点：精确，无边界突发 缺点：内存大（每请求一条记录）

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

近似算法，工业首选：

当前窗口 [now-60, now]
= 当前分钟 已计数 × (已过秒数 / 60) + 上一分钟计数 × (剩余比例)

例：当前 1:30，已计 50；上一分钟（0:00-0:59）计 80
counter = 50 + 80 × (60-30)/60 = 50 + 40 = 90

-- Redis Lua（原子）
local current = redis.call('GET', KEYS[1]) or 0
local previous = redis.call('GET', KEYS[2]) or 0
local elapsed = tonumber(ARGV[1])   -- 当前窗口已过秒数
local rate = tonumber(current) + tonumber(previous) * (60 - elapsed) / 60
if rate >= tonumber(ARGV[2]) then return 0 end
redis.call('INCR', KEYS[1])
redis.call('EXPIRE', KEYS[1], 120)
return 1

优点：精度近 SW Log + 内存仅两个计数器 缺点：近似算法，假设窗口内请求均匀分布

4. Token Bucket（令牌桶）

桶容量 100 个 token
每秒补充 10 个（最大 100）
每个请求消耗 1 个 token
桶空时拒绝

def allow(now):
    elapsed = now - last_refill
    tokens = min(capacity, tokens + elapsed * refill_rate)
    last_refill = now
    if tokens >= 1:
        tokens -= 1
        return True
    return False

优点：

• 允许突发（桶满时可一次发 100 个）
• 长期速率稳定
• AWS / GCP / Stripe 等大厂常用

缺点：实现稍复杂（需要存 tokens + last_refill）

5. Leaky Bucket（漏桶）

桶容量 100，固定速率 10/秒"漏出"
请求加入桶（满则拒绝）

def allow(now):
    leaked = (now - last_check) * leak_rate
    bucket = max(0, bucket - leaked)
    last_check = now
    if bucket + 1 <= capacity:
        bucket += 1
        return True
    return False