在现代分布式系统和微服务架构的面试与实际设计中，限流无疑是最基础也是最重要的环节之一。试想一下，如果你的服务突然涌入数倍于平时的流量，而没有相应的保护机制，会发生什么？没错，服务可能会因为数据库连接耗尽或线程阻塞而彻底崩溃。
这时候，限流就成了我们的“防洪大堤”。在这篇文章中，我们将不再停留在表面的概念，而是深入探讨系统设计中最核心的几种限流算法。我们将一起剖析它们的底层原理，对比它们的优劣势，并通过实际的代码示例来看看如何在我们的项目中实现它们。无论你是正在准备系统设计面试，还是正在构建一个高并发的 API 网关，这篇文章都将为你提供从理论到实践的全面指引。
为什么我们需要限流？
在我们深入算法之前，先明确一点：限流不仅仅是为了“拒绝请求”。它的核心目的是保护系统资源的可用性，同时确保公平性。
通常，我们会遇到两种主要的限流场景：
- 服务端保护：防止后端数据库、缓存或核心服务被突发流量压垮。例如，防止恶意爬虫抓取数据。
- 客户端/合作伙伴限制：在 SaaS 产品中，我们通常需要根据用户的订阅等级（如免费版 vs 企业版）来限制其 API 调用次数（例如：每分钟 100 次）。
在系统设计中，实现限流的位置也很有讲究。你可以在反向代理层（如 Nginx）做，可以在应用网关（如 Zuul, Spring Cloud Gateway）做，甚至可以在单体应用的拦截器中做。分布式系统中，为了保证一致性，我们通常借助 Redis 等中间件来存储计数器状态。
1. 令牌桶算法
令牌桶是业界应用最广泛的限流算法之一。它的核心思想非常直观：系统以恒定的速率向桶中放入“令牌”，而请求处理时需要从桶中拿走令牌。
它是如何工作的？
想象一下，有一个管理员（后台线程）每秒向桶里扔 10 个令牌。当用户请求到来时，我们首先检查桶里有没有令牌。如果有，就拿走一个并处理请求；如果桶空了，那就直接拒绝请求。这里有一个关键的点：桶是有容量的。如果请求量很少，桶里的令牌会累积起来直到装满。这时，如果突然来了一波爆发流量，桶里积累的令牌可以瞬间处理这批突发请求，直到令牌再次耗尽。这就是令牌桶能够应对“突发流量”的原因。
#### 实战代码解析
让我们用 Python 来写一个简单但完整的令牌桶实现。请仔细看代码中的注释，理解令牌补充的逻辑。
import time
class TokenBucket:
def __init__(self, rate, capacity):
# rate: 每秒生成的令牌数
self.rate = rate
# capacity: 桶的最大容量（令牌数上限）
self.capacity = capacity
# 当前令牌数，初始时填满桶
self.tokens = capacity
# 上次补充令牌的时间戳
self.last_refill = time.time()
def allow_request(self):
"""
判断是否允许请求
返回 True: 允许
返回 False: 拒绝
"""
now = time.time()
# 计算距离上次补充过去了多少秒
elapsed = now - self.last_refill
# 根据时间差计算应该补充多少令牌
# 例如：rate=10/s，过了 0.5s，就应该补充 5 个
new_tokens = elapsed * self.rate
# 更新桶内的令牌数，但不能超过容量上限
self.tokens = min(self.tokens + new_tokens, self.capacity)
# 更新最后补充时间
self.last_refill = now
# 检查是否有足够的令牌（这里假设每个请求消耗1个令牌）
if self.tokens >= 1:
self.tokens -= 1
return True
else:
return False
# --- 测试代码 ---
# 设置：每秒生成 5 个令牌，桶最大容量为 10
limiter = TokenBucket(rate=5, capacity=10)
# 模拟突发流量：瞬间发送 15 个请求
allowed_count = 0
for i in range(15):
if limiter.allow_request():
allowed_count += 1
print(f"请求 {i+1}: 允许")
else:
print(f"请求 {i+1}: 拒绝")
print(f"总共允许的请求数: {allowed_count}")
# 预期结果：前 10 个会被允许（因为初始满了），后 5 个被拒绝
#### 为什么选择令牌桶？
在实际开发中，如果我们希望系统既能严格限制平均速率，又能在流量低谷期“攒人品”（积累令牌）以应对偶尔的流量高峰，令牌桶是最佳选择。例如，API 网关限制用户每分钟只能调用 100 次，但允许在短时间内稍微超过平均值，这就非常适合令牌桶。
挑战与注意事项：
- 精度问题：单机版实现很简单，但在分布式系统中，多个实例共享同一个桶的状态时，需要使用 Redis 的 Lua 脚本来保证原子性，否则在高并发下计算会有误差。
- 时间跳跃：服务器的时间发生回拨时，可能会导致计算出的时间差为负数，引发逻辑错误（生产环境通常通过记录当前时间戳的增量来避免）。
2. 漏桶算法
漏桶算法的设计初衷与令牌桶略有不同。如果令牌桶关注的是“允许通过多少”，漏桶关注的就是“强制输出速率恒定”。
核心机制：
漏桶就像一个底部有漏洞的水桶。
- 请求像水一样流入桶中（请求入队）。
- 桶底以恒定的速率漏水（处理请求）。
- 如果流入太快，桶满了，多余的水就会溢出（拒绝请求）。
实战代码解析（带队列版）
虽然很多现代框架用计数器简化实现漏桶，但其本质是包含一个队列的。下面我们展示一个带有真实处理队列的漏桶实现，这能帮你更好地理解它如何平滑流量。
import time
import threading
class LeakyBucket:
def __init__(self, capacity, outflow_rate):
# capacity: 桶的容量（即队列最大长度）
self.capacity = capacity
# outflow_rate: 漏水速率，即每秒处理多少个请求
self.outflow_rate = outflow_rate
# 当前桶里的水量（排队中的请求数）
self.current_load = 0
# 最后处理请求的时间
self.last_leak_time = time.time()
def allow_request(self, thread_id):
"""
尝试往桶里加水（请求）
"""
self._leak() # 先尝试漏水（处理旧的请求）
if self.current_load 0:
# 减少水位，但不能少于0
# 注意：这里是数学模型上的简化，实际队列通常按个 pop
self.current_load = max(0, self.current_load - processed)
self.last_leak_time = now
print(f" [漏水]: 处理了 {processed:.2f} 个请求，剩余水位 {self.current_load:.2f}")
# --- 测试代码 ---
# 设置：桶容量为 5，处理速率为每秒 1 个
bucket = LeakyBucket(capacity=5, outflow_rate=1)
# 模拟快速并发请求
print("--- 开始并发测试 ---")
for i in range(10):
bucket.allow_request(i)
time.sleep(0.1) # 每0.1秒一个请求，远快于处理速率
#### 漏桶 vs 令牌桶：你应该选哪个？
这是一个经典的面试问题。
- 漏桶的主要优势在于平滑流量。它强制输出速率恒定，非常适合保护数据库这种脆弱的下游服务。它不关心突发流量，即使你有一万个令牌，我也只会按固定速率慢慢处理。它的缺点是，无法应对突发业务需求，即使是短暂的空闲期也无法利用。
- 令牌桶则更加灵活。只要桶里有货，它可以瞬间处理高并发。这使得它更适合面向用户的 API 限流。
3. 固定窗口计数器
除了桶类算法，最简单的莫过于计数器了。固定窗口算法将时间划分为固定大小的窗口（如 1 分钟），并在每个窗口内维护一个计数器。
工作原理：
- 我们将时间轴切分成每分钟一段。
- 每当请求到来，
count++。 - 如果
count > limit，拒绝。 - 当时间跨过下一分钟时，
count重置为 0。
实战代码与“临界突变”问题
虽然实现简单，但它在系统设计中有一个著名的缺陷：边界突变。让我们通过代码来看看这个问题有多严重。
import time
class FixedWindowRateLimiter:
def __init__(self, limit, window_size):
self.limit = limit
self.window_size = window_size # 窗口大小，单位秒
self.counter = 0
self.window_start = time.time()
def allow_request(self):
now = time.time()
# 检查是否已过窗口期
if now - self.window_start >= self.window_size:
# 重置计数器和窗口时间
self.counter = 0
self.window_start = now
if self.counter < self.limit:
self.counter += 1
return True
else:
return False
# --- 场景模拟 ---
# 假设限制：每分钟 100 个请求
limiter = FixedWindowRateLimiter(limit=100, window_size=60)
print("--- 模拟临界问题 ---")
# 场景：在第 59 秒时来了 100 个请求，第 61 秒（下一窗口）来了 100 个请求
# 理论上我们应该限制每 60 秒最多 100 个，但这里实际允许了 200 个请求！
在这个算法中，如果我们允许每分钟 100 个请求。在第 59 秒，用户瞬间发了 100 个请求，刚好填满配额。时间刚过 1 秒进入第 61 秒（新窗口），窗口重置，用户又可以瞬间发 100 个请求。结果就是在 2 秒内处理了 200 个请求，这可能直接击垮数据库。这就是固定窗口算法的致命伤。
4. 滑动窗口日志
为了解决固定窗口的“临界突变”问题，滑动窗口算法应运而生。我们可以通过滑动窗口日志来实现。
核心思想：
不再重置计数器，而是记录每一次请求的时间戳。每次新请求到来时，我们检查时间戳队列，移除所有不在当前时间窗口内的旧记录，然后统计剩余记录数量。
import time
from collections import deque
class SlidingWindowLog:
def __init__(self, limit, window_size):
self.limit = limit
self.window_size = window_size
# 使用双端队列存储请求的时间戳
self.request_timestamps = deque()
def allow_request(self):
now = time.time()
# 1. 清理窗口外的过期时间戳（这是滑动的核心）
while self.request_timestamps and self.request_timestamps[0] <= now - self.window_size:
self.request_timestamps.popleft()
# 2. 判断当前窗口内请求数是否超限
if len(self.request_timestamps) < self.limit:
self.request_timestamps.append(now)
return True
else:
return False
# --- 测试代码 ---
limiter = SlidingWindowLog(limit=3, window_size=1)
# 模拟请求
for i in range(5):
if limiter.allow_request():
print(f"Request {i+1}: Allowed")
else:
print(f"Request {i+1}: Rejected")
time.sleep(0.3) # 控制节奏
优缺点分析：
- 优势：精度极高，完美解决了临界突变问题，能够非常平滑地限制流量。
- 挑战：内存开销大。如果限流阈值是 100 万 QPS，我们就需要在内存（或 Redis）中存储 100 万条时间戳记录。这在极端高并发下对性能和内存都是考验。
5. 滑动窗口计数器
这是“固定窗口”和“滑动窗口”的折衷方案，也是很多高性能限流组件（如 Redisson 的 RRateLimiter）采用的实现方式。它不记录每一个时间戳，而是将一个大的时间窗口切分成多个小格。
例如，1 分钟的窗口切分为 6 个 10 秒的小窗口。
- 计数：每个小窗口有自己的计数器。
- 滑动：统计总请求数时，根据当前时间在不同小窗口的占比，加权计算总和。
这种方式在精度和性能之间取得了完美的平衡，既避免了固定窗口的突变，又不像日志算法那样消耗巨大内存。
总结与最佳实践
在这篇文章中，我们一起深入探讨了四种核心的限流算法。让我们来回顾一下，当你下次在设计系统时，应该如何选择：
- 令牌桶：这是你的默认选择。它灵活，能处理突发流量，且实现相对简单。适合 API 网关、微服务调用。
- 漏桶：当你需要强制恒定速率保护下游服务（如数据库、第三方 API）时使用。它能平滑流量，但无法处理突发。
- 固定窗口：仅在简单的、对精度要求不高的场景下使用，或者作为非核心功能的简易实现。要小心临界突变问题。
- 滑动窗口：适合对精度要求极高的业务场景，如计费系统、核心交易风控。要注意内存和 Redis 的开销。
给你的建议：
在实际工程中，不要试图从头手写一个分布式的限流器，除非是为了学习。在生产环境中，强烈建议使用成熟的中间件或库，例如配合 Redis + Lua 脚本 来实现原子性的令牌桶或滑动窗口。这不仅利用了 Redis 的高性能，还解决了分布式环境下的并发一致性问题。
希望这篇文章能帮助你建立起对限流算法的直观理解。系统设计不仅仅是堆砌技术，更是在各种约束条件下寻找最优解的艺术。下次面试时，当面试官问你“如何设计一个秒杀系统的限流”，你就可以自信地说：“我们可以使用滑动窗口算法来防止流量突刺，并结合令牌桶在网关层进行初步拦截……”
祝你编码愉快！