限流算法实测:固定窗口 2 倍溢出、漏桶排队、令牌桶突刺——四种算法到底选哪个
适用读者:给接口加过限流、但说不清"为什么选这个算法"的后端工程师
技术栈:Python 纯标准库实现 + 真实压测;算法本身与语言无关,Django / Spring Boot / Go 通用
阅读时间:约 28 分钟 | 难度:中高级 | 收获:四种算法的可运行实现 + 边界条件实测数据 + 选型决策树
前言:一次被"整点突刺"打挂的接口
线上有个查询接口,加了限流,配置是"每分钟 600 次"。上线后风平浪静,直到某天早上 9:00:00,监控告警:CPU 打满,接口 502。
翻了日志才发现规律:这个接口被一个报表系统定时调用,所有客户端都配置在
整分钟触发。于是每分钟的前 0.2 秒,600 个请求全部涌进来,剩下的 59.8 秒一个请求都没有。
限流器认为:"这一分钟才 600 次,没超啊。" 但你的服务在同一瞬间承受了 600 QPS——
限流阈值是对的,但时间分布完全失控。
这就是
固定窗口计数器的经典缺陷:窗口边界处可以产生 2 倍溢出。而这只是四种主流限流算法里最简单的一种。
这篇文章不满足于"背诵四种算法的定义",而是把每种算法
真的实现出来压测一遍,用数据说明它们的差异:
- 固定窗口为什么在边界处能放过 2 倍流量?
- 滑动窗口日志明明最精确,为什么生产环境反而不敢用?
- 漏桶和令牌桶看起来很像,那个"桶"到底装的是请求还是令牌?
- 令牌桶的"突刺"能力是优点还是隐患?
目录
- 限流到底在解决什么问题
- 固定窗口计数器:最简单,也最容易翻车
- 滑动窗口日志:最精确,也最占内存
- 滑动窗口计数:精度与内存的折中
- 漏桶:恒定输出的"整流器"
- 令牌桶:允许突发的主流选择
- 四种算法横向实测
- 分布式限流的三个坑
- 选型决策树与参数估算
- 上线 Checklist + 高频 FAQ
1. 限流到底在解决什么问题
先分清三件常被混为一谈的事:
限流的本质是一道
准入控制:在请求真正消耗资源之前,决定它能不能进来。它的价值有三层:
- 自我保护:防止单个调用方(或爬虫)耗尽你的连接池、CPU、数据库连接。
- 成本控制:下游是按调用量计费的(大模型 API、短信、地图服务),限流就是控预算。
- 公平性:多租户系统里,不能让一个租户占满所有配额。
限流器通常放在哪?从外到内:
客户端 → CDN/WAF → 网关(Nginx/APIGW) → 应用内限流 → 数据库/下游
↑ 全局限流 ↑ 路由级限流 ↑ 业务级限流
关键认知:限流的位置越靠外,成本越低。 在 Nginx 层拦掉一个请求只花几微秒,等请求进了应用、占了线程、查了数据库再拦,成本已经付出去了。所以理想结构是"多层限流",而不是在应用里堆一个万能限流器。
另外要明确一个概念:
限流的单位通常不是"次/分钟",而是"令牌/秒 + 桶容量"。这句话现在看没感觉,读完令牌桶那一节你就懂了。
2. 固定窗口计数器:最简单,也最容易翻车
原理
把时间切成固定长度的窗口(比如 1 分钟),每个窗口一个计数器。请求来了就 +1,超过阈值就拒绝,窗口到期计数器清零。
import time
class FixedWindowLimiter:
def __init__(self, limit: int, window: float):
self.limit = limit # 窗口内允许的请求数
self.window = window # 窗口长度(秒)
self.count = 0
self.window_start = time.time()
def allow(self) -> bool:
now = time.time()
# 窗口过期,重置
if now - self.window_start >= self.window:
self.window_start = now
self.count = 0
if self.count < self.limit:
self.count += 1
return True
return False
生产环境用 Redis 实现的话,通常是一行
INCR +
EXPIRE:
def allow(user_id, limit=600, window=60):
key = f"rl:{user_id}:{int(time.time() // window)}"
pipe = redis.pipeline()
pipe.incr(key)
pipe.expire(key, window)
count, _ = pipe.execute()
return count <= limit
注意这里 key 里带了 time.time() // window,也就是窗口编号。这样根本不需要手动重置计数器——换窗口就是换 key,旧的自动过期。这是生产中固定窗口最常用的写法。
实测:2 倍溢出
现在来看它最致命的缺陷。配置
limit=10, window=1s,我在第 0.99 秒打 10 个请求,第 1.01 秒再打 10 个请求:
窗口 [0.0s, 1.0s) → 10 个请求 ✓ 全部通过(达到上限)
窗口 [1.0s, 2.0s) → 10 个请求 ✓ 全部通过(新窗口,计数器归零)
结果:0.02 秒内通过了 20 个请求,实际瞬时速率 = 1000 QPS
而配置的阈值是 10 QPS。
溢出倍率 = 2 倍,是理论上限。一般形式是:跨窗口边界时,最多可以通过
2 × limit 个请求。我用假时钟实测确认了这个上限:
limit=10/窗口=1s,两个边界各打 10 个,
20 个全部放行(见第 7 节实测表)。
这就是开头那次事故的原因。报表系统在整分钟触发,恰好卡在每个窗口的起点,所以每个窗口的流量都集中在开头——限流器看到的永远是"刚好达标"。
优缺点
结论:固定窗口只适合"防呆",不适合"防精确超量"。 如果你的下游按量计费,或者 SLA 有严格 QPS 上限,别用它。
3. 滑动窗口日志:最精确,也最占内存
原理
既然问题出在"窗口是死的",那就让窗口
跟着当前时间滑动。记录每个请求的时间戳,判断"最近 1 秒内有多少个请求"。
import time
from collections import deque
class SlidingWindowLogLimiter:
def __init__(self, limit: int, window: float):
self.limit = limit
self.window = window
self.log = deque() # 存请求时间戳
def allow(self) -> bool:
now = time.time()
# 移除窗口外的过期记录
while self.log and now - self.log[0] >= self.window:
self.log.popleft()
if len(self.log) < self.limit:
self.log.append(now)
return True
return False
实测:精度无可挑剔
同样的场景,
limit=10, window=1s,在第 0.99 秒打 10 个请求,第 1.01 秒再打 10 个:
0.99s: 窗口内记录 0 条 → 前 10 个全部通过 → 日志 = [0.99 × 10]
1.01s: 窗口回溯到 -0.01s,0.99s 的 10 条记录仍在窗口内
→ 第 11 个请求被拒绝 ✓(len(log) = 10 已达上限)
没有任何边界漏洞。 任意时刻向前看 1 秒,通过的请求数都不会超过 10。
代价:内存与复杂度
精确是有代价的:
单用户单接口就要这么多。如果按 user_id × API 维度限流,内存会被迅速吃光。这就是"精确但不敢用"的原因。
Redis 里用
ZADD +
ZREMRANGEBYSCORE 可以实现,但每次请求都是 3~4 条命令,且
ZSET 的成员数就是 QPS × 窗口秒数。有人用
member = 时间戳:随机数 来避免同毫秒冲突,进一步放大了内存。
优缺点
结论:滑动窗口日志适合低频、高价值的限流——比如"每分钟最多 5 次验证码",因为 5 × 60 = 300 条记录,内存完全无所谓。
4. 滑动窗口计数:精度与内存的折中
原理
日志版太占内存,是因为它记了每个请求的
精确时刻。但其实我们只需要知道"大概有多少个"。于是把窗口切成 N 个小格子(比如 1 分钟切成 6 个 10 秒的格子),每个格子存一个计数。
估算当前窗口的请求数时,把所有还在窗口内的格子按
重叠比例加权求和:
当前窗口请求数 ≈ Σ (每个格子内的计数 × 该格子与窗口的重叠比例)
import time
class SlidingWindowCounter:
def __init__(self, limit: int, window: float, buckets: int = 6):
self.limit = limit
self.window = window
self.bucket_size = window / buckets
self.n = buckets
self.counts = {} # bucket_id -> count
def _bucket(self, ts):
return int(ts // self.bucket_size)
def allow(self) -> bool:
now = time.time()
current = self._bucket(now)
# 清理已完全滑出窗口的桶(保留 current-n,它可能还有部分重叠)
for b in list(self.counts):
if b < current - self.n:
del self.counts[b]
# 最老的桶只保留重叠部分,其余整桶计入
elapsed = (now % self.bucket_size) / self.bucket_size
estimate = 0.0
for b, c in self.counts.items():
weight = max(0.0, 1.0 - elapsed) if b == current - self.n else 1.0
estimate += c * weight
if estimate < self.limit:
self.counts[current] = self.counts.get(current, 0) + 1
return True
return False
一个真实的踩坑:只估一个桶 = 完全失效
我第一次实现时,只拿"上一个桶 × 重叠比例 + 当前桶"来估算——看起来像那么回事,实测却放过了
600 个请求(误差 500%)。原因很简单:窗口是 1 秒、有 6 个桶,而我每个桶都只回溯一个桶,等于把 1 秒的窗口缩成了 0.167 秒。
必须遍历窗口内的所有桶,只有"即将滑出的那个最老桶"才需要按比例打折。
# ❌ 错误:只回溯一个桶,等价于把窗口缩短了
estimate = prev * (1 - elapsed) + curr
# ✅ 正确:遍历窗口内所有桶,只对最老的桶打折
for b, c in self.counts.items():
weight = max(0.0, 1.0 - elapsed) if b == current - self.n else 1.0
estimate += c * weight
修正后实测:均匀 1000 请求/秒的流量下,放行数精确为
100(理论值 100),误差
0%。
精度误差
滑动窗口计数是
近似算法,误差来源只有一个:它假设"上一个窗口内的请求是均匀分布的"。实际当然不是——如果请求全部集中在一个桶的末尾,估算就会偏离。
不过只要桶数量足够(6~10 个),误差通常控制在个位数百分比以内,相比固定窗口的
100% 溢出是质的改善。桶越多越精确,代价是遍历和内存略增。
优缺点
结论:如果你的场景不极端,滑动窗口计数是最平衡的方案。 多数云厂商 API 网关的默认限流就是这个。
5. 漏桶:恒定输出的"整流器"
原理
想象一个底部有洞的桶,水(请求)从上面倒进来,从洞(出口)以
恒定速率流出去。
- 桶满了 → 新来的水溢出(拒绝请求)
- 桶没满 → 水先存着,排队等待流出
import time
class LeakyBucketLimiter:
def __init__(self, capacity: int, leak_rate: float):
self.capacity = capacity # 桶容量
self.leak_rate = leak_rate # 每秒漏出(处理)多少请求
self.water = 0.0
self.last_leak = time.time()
def allow(self) -> bool:
now = time.time()
# 先漏水:按流逝时间计算漏出量
leaked = (now - self.last_leak) * self.leak_rate
self.water = max(0.0, self.water - leaked)
self.last_leak = now
if self.water + 1 <= self.capacity:
self.water += 1
return True
return False # 桶满,溢出(拒绝或排队)
关键点:漏桶的核心是"输出速率恒定"。 稳态下,无论输入多么汹涌,出口永远以
leak_rate 的速度放行。
一个容易被忽略的细节:冷启动突刺
上面这段代码(以及绝大多数教学实现)有一个隐藏行为:
桶初始是空的。所以系统刚启动、或者长时间空闲之后,第一批
capacity 个请求会被
立刻放行——它们填满桶的瞬间也在以
leak_rate 漏出。
实测
capacity=100, leak_rate=100, 输入 1000 请求/秒,逐 0.1 秒统计放行数:
[0.0s - 0.1s) 100 个 ← 冷启动瞬间放行,等于 capacity
[0.1s - 0.2s) 19 个 ← 开始收敛
[0.2s - 0.3s) 10 个 ┐
[0.3s - 0.4s) 10 个 │
... ├─ 稳态:严格 100/s
[0.9s - 1.0s) 10 个 ┘
1 秒总计 199 个(100 冷启动 + 99 稳态),
0.2 秒之后实测速率 = 100.0/s,与 leak_rate 完全一致。
这告诉我们两件事:
- 漏桶的"恒定"是稳态性质,不是瞬态性质。 冷启动/空闲后的第一个瞬间仍可能有一波
capacity 大小的流量。如果你的下游连这一波都扛不住,把 capacity 调小。
- 它和令牌桶的突刺在数学上其实是同一回事——初始桶满(令牌桶)或初始桶空(漏桶),都会产生一次
capacity 量级的瞬态。区别在于漏桶的突刺只发生在冷启动,令牌桶则每次空闲后都会重新积攒。
漏桶 vs 令牌桶:一个字的区别
这是最容易混淆的地方。两者的"桶"装的东西完全不同:
一句话总结:
漏桶管"出场速度",令牌桶管"平均速度 + 允许突刺"。
优缺点
结论:漏桶适合"下游必须被温柔对待"的场景,但要记得把
capacity 调小以压住冷启动突刺。另外它带来的排队延迟要警惕——如果客户端有超时限制,排队请求可能会超时,这时宁可拒绝也不要无脑排队。
6. 令牌桶:允许突发的主流选择
原理
反过来想:桶里不放请求,放
令牌。系统以恒定速率往桶里扔令牌,桶满则令牌溢出。请求进来时,必须先拿走一个令牌才能通过。
import time
class TokenBucketLimiter:
def __init__(self, rate: float, capacity: int):
self.rate = rate # 令牌生成速率(个/秒)
self.capacity = capacity # 桶容量(最大突发量)
self.tokens = float(capacity)
self.last_refill = time.time()
def allow(self, tokens_needed: int = 1) -> bool:
now = time.time()
# 补充令牌
self.tokens = min(
self.capacity,
self.tokens + (now - self.last_refill) * self.rate
)
self.last_refill = now
if self.tokens >= tokens_needed:
self.tokens -= tokens_needed
return True
return False
"突刺"从哪来
因为桶
可以积攒令牌。如果系统空闲了 10 秒,桶里就攒了
10 × rate 个令牌(上限是 capacity)。此时突然来了一批请求,可以
一次性消费掉所有积攒的令牌,这就是突发(burst)。
举个例子,
rate = 10/s, capacity = 100:
空闲 10 秒后,桶里有 min(100, 10×10) = 100 个令牌
→ 第一个瞬间可以放行 100 个请求(10 倍于平均速率)
→ 之后被限制回 10/s
这是一个设计选择,不是 bug。 真实流量本来就不是均匀的——用户早上打开 App,会一次性加载十几个接口。如果严格按 10/s 限流,这些正常用户会被误伤。令牌桶的 capacity 就是用来"吸收正常突刺"的。
但要小心:
capacity 设得太大 = 变相放开限流。如果
capacity = 10000, rate = 10,那么理论上攻击者可以每 1000 秒打一次、每次打 10000 个请求,平均速率还是 10/s,但瞬时峰值会让你的下游崩溃。
Redis + Lua 生产实现
生产环境的令牌桶必须用 Lua 保证原子性(读-算-写三步不能被打断):
-- KEYS[1]: 桶的 key
-- ARGV[1]: rate(令牌/秒)
-- ARGV[2]: capacity(桶容量)
-- ARGV[3]: now(毫秒时间戳)
-- ARGV[4]: requested(本次要消耗的令牌数)
local key = KEYS[1]
local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])
local data = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(data[1]) or capacity
local last_refill = tonumber(data[2]) or now
-- 按流逝时间补充令牌
local delta = math.max(0, now - last_refill)
tokens = math.min(capacity, tokens + delta * rate / 1000)
local allowed = tokens >= requested
if allowed then
tokens = tokens - requested
end
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
-- TTL 设为"装满一桶所需时间的 2 倍",避免冷 key 常驻内存
redis.call('EXPIRE', key, math.ceil(capacity / rate * 2))
return allowed and 1 or 0
为什么要用 Lua:如果分成
HGET → 计算 →
HSET 三步,两个并发请求可能读到同一个令牌数,然后各自扣减,导致令牌被超发——这正是幂等文章里说的 TOCTOU 问题。
优缺点
结论:令牌桶是通用推荐的默认选择。 它同时给了你"平均速率"和"突发容量"两个旋钮,表达能力最强。
7. 四种算法横向实测
所有数据来自
tools/bench_rate_limit.py(纯标准库 + 假时钟,可复现)。统一参数:
limit = 100 请求/秒,模拟 1000 个请求在 1 秒内均匀到达。
实测结论,逐条对应文章前面的分析:
- 固定窗口的边界溢出被实测确认:
limit=10/窗口=1s,在 0.99s 打 10 个、1.01s 再打 10 个,20 个全部通过,恰好 2 倍。这是理论上限,实测完全吻合。
- 滑动窗口日志精确无误:同样场景只放行 10 个,第 11 个被正确拒绝。
- 修正后的滑动窗口计数达到 0% 误差:1000 请求/秒的均匀流量下放行 100 个,与理论值完全一致。而修正前的错误实现放行 600 个(500% 误差)——这就是第 4 节那个"只回溯一个桶"的坑。
- 令牌桶的突刺实测 = 100 个(初始桶满),加稳态 100/s,1 秒共 199 个。
- 令牌桶排空后(
tokens = 0)不再有突刺:放行 99 个,退化为纯速率限制。这验证了"突刺完全由 capacity 决定"的判断。
- 漏桶冷启动突刺同样 = 100 个:初始空桶被瞬间填满,之后严格 100/s。稳态速率实测 100.0/s,与
leak_rate 分毫不差。
把 4 和 6 放在一起看:令牌桶和漏桶在"突刺"这件事上行为一致,区别只是令牌桶每次空闲都会重新积攒,漏桶只有冷启动那一次。
同样的参数,看时间维度
固定窗口(limit=100/s,窗口不滑动)
0.0s ──────── 100 通过 ──────── 0.99s │ 1.0s ──── 100 通过 ──── 1.99s
↑
边界处 20ms 内过 200 个 ⚠️
令牌桶(rate=100/s,capacity=100)
0.0s ──── 100 通过(桶内积攒)──── 0.01s │ ──── 100/s 匀速 ────
↑ 突刺 = capacity,空闲后会重新积攒
漏桶(rate=100/s,capacity=100)
0.0s ──── 100 通过(冷启动填桶)── 0.1s │ ──── 严格 100/s ────
↑ 突刺 = capacity,仅此一次
8. 分布式限流的三个坑
前面都是单机实现,一旦上多实例,问题立刻出现。
坑一:本地限流 × N 实例 = 阈值 × N
你在 4 台机器上用本地令牌桶限流 100 QPS,以为整体是 100 QPS,实际是
400 QPS。Nginx 的
limit_req 在单机生效,多实例部署时同样有此问题。
解法:要么在网关层做全局限流(推荐),要么用 Redis 做集中计数。
坑二:Redis 成了新的瓶颈和单点
所有请求都要打一次 Redis,Redis 的 QPS 上限反而成了系统上限。
解法:
-
本地 + 全局两层:本地令牌桶吸收大部分流量(阈值设为全局的 1/N 加缓冲),只有本地判断接近阈值时才去问 Redis。这样 Redis 的调用量下降 90% 以上。
-
批量同步:本地按固定间隔(如每 100ms)从 Redis 批量领取一批令牌,而不是每个请求都问一次。
- Redis 部署为集群/哨兵,避免单点。
坑三:限流 key 的设计
key 的粒度决定了限流的语义:
要小心的两个陷阱:
- IP 限流会误伤 NAT 用户。一个公司出口 IP 后面可能有几千人,按 IP 限流会让整个公司一起被拒。可以改用
user_id 或 user_id + ip 组合。
- key 爆炸。如果 key 里带了不可枚举的维度(如 URL 查询参数),攻击者可以构造海量不同 key,绕过限流同时把 Redis 撑爆。key 的维度必须来自可控白名单。
9. 选型决策树与参数估算
决策树
需要极严格的精确性(如验证码、支付)?
├─ 是 → 滑动窗口日志(低频场景,内存无所谓)
└─ 否 ↓
下游脆弱、必须恒定速率?
├─ 是 → 漏桶
└─ 否 ↓
需要允许正常突发(如 App 启动批量请求)?
├─ 是 → 令牌桶 ★ 推荐默认
└─ 否 ↓
一般场景 → 滑动窗口计数 ★ 性价比最高
参数估算:怎么定 rate 和 capacity
这是实践中最难回答的问题。推荐步骤:
- 先测出下游容量。在压测环境用递增并发找出 P99 延迟开始劣化的拐点,那就是
rate 的上限。不要拍脑袋定值。
- rate = 下游容量的 70%~80%,留出余量应对突发的非限流流量。
- capacity(令牌桶)= 单次正常突发的请求数。观察真实客户端行为:一个 App 首页启动会打几个接口?如果是 15 个,那 capacity 至少 15。
- 拒绝策略要明确:返回
429 Too Many Requests,并带上 Retry-After 头。
HTTP/1.1 429 Too Many Requests
Retry-After: 3
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1726450000
把限流信息透出给客户端,是专业度的体现——好的客户端会读
Retry-After 并退避重试,而不是疯狂重试加剧拥塞。
一个常见的误配置
# ❌ 错误:以为这是"每用户 600 次/分钟"
key = f"rl:{user_id}" # 没带时间维度
INCR key; EXPIRE key 60
# 问题:第一次 EXPIRE 后,如果持续有请求,key 永不过期
# → 变成"累计 600 次后永久封禁"
正确写法是让 key 自带时间维度(固定窗口),或用 Lua 维护滑动窗口的状态。
10. 上线 Checklist + 高频 FAQ
Checklist
- [ ] 限流阈值基于压测数据,不是拍脑袋
- [ ] 限流 key 的维度来自白名单,防止 key 爆炸绕过
- [ ] 多实例部署时,确认是全局限流而非本地限流 × N
- [ ] Redis 令牌桶用 Lua 脚本保证原子性
- [ ] Redis key 设置了 TTL,避免冷 key 常驻
- [ ] 返回
429 且带 Retry-After / X-RateLimit-* 响应头
- [ ] 被限流的请求有监控和告警(突然大量 429 说明阈值或流量异常)
- [ ] 限流器本身有降级方案:Redis 挂了怎么办?通常选择放行(fail-open)而不是全部拒绝(fail-close),避免限流组件成为新的单点
- [ ] 内部服务间调用、健康检查、监控探针等流量加入白名单,否则会把运维通道一起限掉
FAQ
Q:限流和服务器的连接数限制有什么区别?
A:连接数限制在 TCP 层,管的是"同时有多少个连接"(并发数);限流在应用层,管的是"单位时间允许多少个请求"(速率)。两者互补:一个防连接堆积,一个防请求洪峰。
Q:为什么我的限流看起来不生效?
A:最常见三个原因——(1) 多实例部署导致阈值被放大 N 倍;(2) 客户端走的是不同的入口,绕过了限流器;(3) key 设计有问题,每个请求都被算成了新用户。先用日志把实际参与限流的 key 打出来排查。
Q:令牌桶的 capacity 该设多大?
A:等于"你希望容忍的最大突发请求数"。可以从观察真实客户端的突发行开始——比如 App 首页并发 15 个请求,就设 30(留一倍余量)。设得太大就等于没有限流。
Q:Redis 挂了限流软件该怎么办?
A:默认
fail-open(放行)。理由:限流的目的是保护服务,如果限流组件本身故障就让全部业务不可用,损失远大于短时限流失效。但要有告警,且对于按量计费的敏感接口可以考虑 fail-close。
Q:Nginx limit_req 和 limit_conn 用哪个?
A:
limit_req 是漏桶算法管速率,
limit_conn 管并发连接数。两个都要配,因为它们是正交的——一个接口可以 QPS 很低但每个请求都很慢,把连接占满。
Q:限流和降级应该谁先触发?
A:限流在前。流量先被削峰,如果削峰后下游仍然过载(说明是内部问题而非流量问题),才触发熔断降级。顺序错了会导致正常流量被误降级。
小结
四种算法的本质区别,可以归结为一个问题:
你想控制的是"平均速率"还是"瞬时速率"?
如果只记一句话:
默认选令牌桶,参数不极端时用滑动窗口计数,验证码这类低频高价值校验用滑动窗口日志。 固定窗口只在无关紧要的地方用。
还有一个贯穿全文的反直觉结论:
漏桶并不比令牌桶"更平滑"——在突刺这件事上,两者都会在初始瞬间放出
capacity 个请求,区别只是令牌桶每次空闲后重新积攒、漏桶只有冷启动那一次。真正决定瞬时峰值的,是你设的
capacity,而不是算法的名字。
限流不是加个装饰器就完事的配置项,它是一道需要压测验证、需要监控、需要考虑失效边界的
系统能力。想清楚"为什么是这个算法、这个参数",比会写代码重要得多。
本文的四种算法实现与全部实测数据均可复现:算法代码是纯 Python 标准库(无第三方依赖),压测脚本在 tools/bench_rate_limit.py(基于假时钟,结果确定可重复)。生产环境请用 Redis + Lua 版本替换单机实现。