限流算法

可视化演示

模拟令牌桶、漏桶与滑动窗口三类限流算法的请求处理过程，对比突发流量接纳与匀速输出特性，观察拒绝与放行判定。

限流概念：为什么池子必须设上限

温泉池的安全线是“一百人”。老板发现，问题不在平时的三五成群，而在周末节假日的爆发：有人上午十点就在大门口排了两个小时队，一开门就演变成“百米冲刺”，救生员根本数不过来。如果不设限，池子瞬间超载，轻则体验崩盘，重则出安全事故。限流，就是提前在山路入口处立一道闸，让游客有序、限量地进入。

计算机系统的“温泉池”就是你的后端服务，它的容量由 CPU 核数、内存、数据库连接池共同决定。平时请求平稳，一切安好；一旦遇上秒杀、热点新闻、恶意爬虫，QPS 可以在几秒内翻几十倍。请求打进来本身不致命，致命的是每个请求背后都要查数据库、扣库存、调外部服务——每个请求都会消耗有限的资源。资源耗尽时，所有请求一起超时，系统像多米诺骨牌一样整体倒下，这就是雪崩。

限流（Rate Limiting）就是预先设置一条“安全水位线”：超过水位的请求被拒绝或排队，保证系统永远不超载。和限流并肩作战的还有两个兄弟：熔断（下游故障时快速断开，避免雪崩扩散）和降级（牺牲次要功能、保全核心链路）。评价一个限流方案，主要看三个维度：精度（放行量是否严格不超阈值）、突发容忍度（能否应对短时脉冲）、资源开销（内存与计算成本）。

老板拍板的规则就是阈值，救生员就是执行器。可阈值定多大、用什么规则定，决定了度假村能安稳多久——接下来，三位管家的方案轮流登场。

漏桶算法：匀速传动的“刚性传送带”

老陈的传送带，在计算机里有个学名叫“漏桶”。想象一个底部带小孔的桶：不管你用消防泵往里面灌水，还是拿滴管一滴一滴地喂，桶底的水流速度都恒定。桶有大小，水灌满了就会溢出——溢出的水，就是被限流挡在门外的请求。

在漏桶算法里，每个请求先进入桶（相当于排队），桶底按固定速率放行请求；桶的容量为，请求超过容量后直接溢出丢弃。这里的“溢出”有两种处理策略：直接拒绝（返回 429 或错误码），或者静默丢弃。无论桶里积压了多少，出去的速率始终不变——这就是漏桶最核心的性质：输出绝对平滑。

老陈的传送带“稳定压倒一切”，但代价是死板：就算池子空无一人，传送带也不会加速。漏桶最适合削峰填谷——把突发的流量抹平成均匀的节奏，喂给那些“吃不了快饭”的下游，比如消息队列的消费者。

令牌桶算法：能攒家底的“温泉币”

小周看不上传送带的死板。他设计的温泉币兑换机每秒往箱子里吐一个币，箱子最多攒十个币；游客必须投一个币才能进池。平时没人来，币就攒着；早高峰一到，攒下的十个币可以让十个游客瞬间冲进去，而后面的游客只能等下一个币。这就是“闲时积攒、忙时爆发”。

令牌桶算法（Token Bucket）就是这个机制：系统按固定速率向桶中放入令牌，桶容量为；每个请求必须先拿走一个令牌才能被处理。有令牌立刻放行，没令牌直接拒绝（或短暂等待）。与漏桶最大的区别是：令牌桶没有请求队列，请求之间不会互相排队，只要桶里有令牌就放行。

小周的方案最懂人性：既保证平均速率不超过，又允许短时间爆发个请求。这也是为什么业界最常用令牌桶——Guava 的 RateLimiter、Nginx 的 limit_req、各大 API 网关的限流插件，基本都是令牌桶或它的变体。

滑动窗口：救生员的“滚动记录板”

小林发现两位管家都漏了一个场景：如果只约定“每分钟最多一百人”，游客会在第 59 秒冲进一百人，第 61 秒（下一分钟的第一秒）再冲进一百人——前后不过 2 秒，池子涌进两百人。这就是“固定窗口”的临界突发问题，也叫“窗口边界穿透”。

滑动窗口的解决思路是：把时间切成更细的小格（比如 1 秒一格），维护一个“过去 60 秒”的滚动计数。每过一小格，最旧的一格滚出窗口，新的一格加入统计。无论请求卡在哪一秒，统计的都是“最近 60 秒”，边界漏洞被彻底封死。

精确实现可以用时间戳列表（每个请求记一个时间戳，滑动时剔除过期的），但内存随请求量增长；工程上更常用“加权滑动窗口”，只需两个计数器就能估算：

小林的方案最“较真”：精确但费资源。滑动窗口适合“用户维度精确配额”——比如每个 API Key 每分钟最多 30 次调用。它和令牌桶并不互斥：令牌桶管“总闸”，滑动窗口管“细账”，很多系统两者叠加使用。

参数配置：阈值不是拍脑袋定出来的

三套方案摆在桌上，老板问出了最关键的问题：阈值怎么定？传送带五秒一人还是一秒一人？温泉币箱子容量十个还是五十个？小林说：别猜，先做一次“压力测试”。

压测的做法是：不断加大并发请求，观察系统的吞吐量曲线。一开始吞吐量随并发上升，但到某个点后，再加并发，吞吐量不再增长、响应时间反而直线恶化——这个点就是系统的“拐点”，也就是真实承载上限。限流阈值应该设在拐点之下，预留安全余量：

参数不是定一次就完事，还要灰度调优：先放量观察，再逐步逼近阈值，同时盯着 CPU、响应时间和错误率三个仪表盘。老板听完，决定先试运营一周，再根据周末客流微调传送带速度——这套“压测找边界、余量保安全、数据验证”的流程，既适用于温泉池，也适用于任何高并发系统。

工程对比：三把钥匙各开一把锁

最后老板拍板：不同入口用不同方案——更衣室用传送带（要平滑），温泉池用温泉币（要允许早高峰），园区大门用记录板（要防踩点）。工程上的选型也是同理：先看你的核心诉求，再决定用哪把锁。

三个关键维度对比：

• 突发流量应对：漏桶完全没有突发能力，秒内最多放行个；令牌桶允许先爆发个再匀速；滑动窗口限制的是窗口总量，剩余额度决定了突发空间。

• 时钟漂移容忍：漏桶和令牌桶都依赖“本地时间差”来补令牌或漏水，对时钟漂移不敏感，分布式部署更稳；滑动窗口依赖每个请求的时间戳，各机器时钟不一致会导致统计偏差。

• 分布式实现：令牌桶和漏桶的核心状态只是一个计数器和时间戳，用 Redis + Lua 就能原子完成；滑动窗口需要 ZSET 或分段计数，成本和复杂度更高。

用一条公式总结三种算法在时间内的“最大放行量”：

选型建议一句话：求稳选漏桶，求活选令牌桶，求准选滑动窗口。大型系统从来不只锁一把锁——网关层用令牌桶挡总量，业务层用滑动窗口做用户配额，消息管道用漏桶做削峰填谷，再叠加“本地限流 + 集中式兜底”的双层结构，才能在突发流量面前既稳又准。

AI 应用场景

Python 演示

import timeimport threading  class TokenBucket:    """令牌桶限流器: rate 为每秒令牌数, capacity 为桶容量"""     def __init__(self, rate, capacity):        self.rate = rate        self.capacity = capacity        self.tokens = capacity        self._lock = threading.Lock()        self._last = time.monotonic()     def acquire(self):        """尝试获取一个令牌, 成功返回 True, 失败返回 False"""        with self._lock:            now = time.monotonic()            self.tokens = min(self.capacity,                              self.tokens + (now - self._last) * self.rate)            self._last = now            if self.tokens >= 1:                self.tokens -= 1                return True            return False  class LeakyBucket:    """漏桶限流器: 固定速率匀速放行, capacity 为桶容量"""     def __init__(self, rate, capacity):        self.rate = rate        self.capacity = capacity        self.queue = 0        self._lock = threading.Lock()        self._last = time.monotonic()     def acquire(self):        """尝试进入队列, 队列未满则接收, 满了则溢出拒绝"""        with self._lock:            now = time.monotonic()            self.queue = max(0.0,                             self.queue - (now - self._last) * self.rate)            self._last = now            if self.queue < self.capacity:                self.queue += 1                return True            return False  class SlidingWindow:    """滑动窗口限流器: 记录时间戳列表, 窗口内总数不超过 limit"""     def __init__(self, window, limit):        self.window = window        self.limit = limit        self._stamps = []        self._lock = threading.Lock()     def acquire(self):        """剔除过期时间戳后, 判断窗口内计数是否超限"""        with self._lock:            now = time.monotonic()            self._stamps = [t for t in self._stamps if now - t < self.window]            if len(self._stamps) < self.limit:                self._stamps.append(now)                return True            return False  def run(name, limiter, seconds, qps):    interval = 1.0 / qps    allowed = 0    total = 0    for _ in range(int(seconds * qps)):        total += 1        if limiter.acquire():            allowed += 1        time.sleep(interval)    print(f"{name}: 总请求 {total}, 放行 {allowed}, "          f"拒绝 {total - allowed}, 放行率 {allowed / total:.1%}")  def main():    print("=== 三种限流算法对比: 模拟 1 秒内突发 100 个请求 ===")    run("令牌桶(rate=50, cap=20)",        TokenBucket(rate=50, capacity=20), 1, 100)    run("漏桶(rate=50, cap=20)",        LeakyBucket(rate=50, capacity=20), 1, 100)    run("滑窗(window=1, limit=50)",        SlidingWindow(window=1, limit=50), 1, 100)  if __name__ == "__main__":    main()

1. self.tokens = min(self.capacity, self.tokens + (now - self._last) * self.rate)令牌桶先按流逝时间补发令牌，再封顶到容量，这是令牌桶的核心公式。
2. self.queue = max(0.0, self.queue - (now - self._last) * self.rate)漏桶先把队列按固定速率“漏”掉一部分，模拟匀速出水。
3. if self.queue < self.capacity:漏桶队列没满就接收请求，满了就溢出拒绝，多余的游客只能走人。
4. self._stamps = [t for t in self._stamps if now - t < self.window]滑动窗口先剔除过期时间戳，只保留窗口内的记录。
5. if len(self._stamps) < self.limit:窗口内计数未超限才放行，精确但不允许任何突发。
6. with self._lock:加锁保证多线程并发请求时计数不超卖，相当于救生员“一次只核一个账”。
7. time.monotonic()使用单调时钟，不受系统改时间影响，避免时钟回拨导致误放行。

常见误区

面试题