系统设计：高并发限流器
当系统面临突发流量或恶意请求时，限流器是保护服务稳定性的第一道防线。
为什么需要限流？
典型场景
| 场景 | 风险 | 限流策略 |
|---|---|---|
| 秒杀活动 | 瞬间百万级请求压垮数据库 | 令牌桶限流 + 队列削峰 |
| API 接口开放 | 被爬虫/刷量 | 按用户/IP 限流 |
| 慢查询接口 | 单个请求耗时长，拖垮线程池 | 并发数限流 |
| 第三方服务调用 | 超调用配额导致被封禁 | 固定窗口计数器 |
限流的核心维度
- 时间维度：每秒/每分允许多少请求
- 资源维度：并发连接数、CPU/内存使用率
- 用户维度：按用户 ID、IP、API Key 区分限流
四种限流算法
算法一：计数器（固定窗口）
最简单的限流算法，单位时间内统计请求次数。
import time
class FixedWindowRateLimiter:
def __init__(self, limit: int, window_seconds: int):
self.limit = limit
self.window = window_seconds
self.current_count = 0
self.window_start = time.time()
self.lock = threading.Lock()
def allow(self) -> bool:
with self.lock:
now = time.time()
if now - self.window_start >= self.window:
self.window_start = now
self.current_count = 0
if self.current_count < self.limit:
self.current_count += 1
return True
return False
缺点：边界突刺问题
窗口1: |████████████| 窗口2: |████████████|
[0s 60s) [60s 120s)
100/100 在 55-60s 发完 100/100 在 60-65s 发完
=> 55-65s 这 10 秒内实际通过了 200 个请求！
算法二：滑动窗口
将固定窗口细分，统计最近 N 个子窗口的请求总数。
import time
from collections import deque
import threading
class SlidingWindowRateLimiter:
def __init__(self, limit: int, window_seconds: int, granulity: int = 10):
"""
granulity: 将窗口划分为多少个子窗口
"""
self.limit = limit
self.window = window_seconds
self.sub_window = window_seconds / granulity
self.granulity = granulity
self.counts = deque(maxlen=granulity) # [(timestamp, count)]
self.lock = threading.Lock()
def allow(self) -> bool:
with self.lock:
now = time.time()
# 清理过期的子窗口
cutoff = now - self.window
while self.counts and self.counts[0][0] <= cutoff:
self.counts.popleft()
current_total = sum(c for _, c in self.counts)
if current_total < self.limit:
# 在当前子窗口计数
if self.counts and now - self.counts[-1][0] < self.sub_window:
self.counts[-1] = (self.counts[-1][0], self.counts[-1][1] + 1)
else:
self.counts.append((now, 1))
return True
return False
优点：平滑了固定窗口的突刺问题
缺点：需要维护多个子窗口的计数，内存占用稍高
算法三：漏桶（Leaky Bucket）
想象一个底部有洞的桶，请求像水一样进入桶，以固定速率流出处理。
import time
import threading
class LeakyBucketRateLimiter:
def __init__(self, rate: float, capacity: int):
"""
rate: 每秒流出速率（个/秒）
capacity: 桶容量
"""
self.rate = rate
self.capacity = capacity
self.water = 0.0
self.last_time = time.time()
self.lock = threading.Lock()
def allow(self) -> bool:
with self.lock:
now = time.time()
# 计算这段时间流出的水量
leaked = (now - self.last_time) * self.rate
self.water = max(0, self.water - leaked)
self.last_time = now
if self.water < self.capacity:
self.water += 1
return True
return False
特点：
- 输出速率绝对均匀（适合需要严格控制下游速率的场景）
- 如果流量突增但桶满了，会直接拒绝，不缓存请求
算法四：令牌桶（Token Bucket）
以固定速率向桶中放入令牌，请求需要拿到令牌才能通过。
import time
import threading
class TokenBucketRateLimiter:
def __init__(self, rate: float, capacity: int):
"""
rate: 每秒放入令牌数
capacity: 桶容量（最大突发流量）
"""
self.rate = rate
self.capacity = capacity
self.tokens = float(capacity) # 初始满桶
self.last_time = time.time()
self.lock = threading.Lock()
def allow(self, tokens_needed: int = 1) -> bool:
with self.lock:
now = time.time()
# 补充令牌
self.tokens = min(
self.capacity,
self.tokens + (now - self.last_time) * self.rate
)
self.last_time = now
if self.tokens >= tokens_needed:
self.tokens -= tokens_needed
return True
return False
特点：
- 允许一定突发流量（桶内积累的令牌）
- 长期来看速率不超过设定值
- 业界最常用：Guava RateLimiter、Nginx limit_req 都是令牌桶
四算法对比
| 算法 | 突发流量 | 平滑输出 | 内存开销 | 实现复杂度 | 适用场景 |
|---|---|---|---|---|---|
| 固定窗口 | 边界突刺 | ❌ | 一个计数器 | 简单 | 简单统计 |
| 滑动窗口 | ✅ 有限 | ✅ | 多个子窗口 | 中等 | 通用限流 |
| 漏桶 | ❌ | ✅ 绝对平滑 | 一个浮点数 | 中等 | 严格限速（如外部 API） |
| 令牌桶 | ✅ 有突发上限 | ✅ 长期平滑 | 一个浮点数 | 中等 | 最常用 |
分布式限流：Redis + Lua
单机限流在分布式系统中不够用，需要共享计数器。Redis 是最佳选择。
固定窗口 Redis 实现
-- rate_limit_fixed.lua
-- KEYS[1]: 限流 key（如 rate_limit:api:/order:user_123）
-- ARGV[1]: 窗口大小（秒）
-- ARGV[2]: 限制次数
local key = KEYS[1]
local window = tonumber(ARGV[1])
local limit = tonumber(ARGV[2])
local current = redis.call('GET', key)
if current == false then
current = 0
else
current = tonumber(current)
end
if current >= limit then
return 0 -- 拒绝
end
-- 首次设置时加过期时间
if current == 0 then
redis.call('SET', key, 1, 'EX', window)
else
redis.call('INCR', key)
end
return 1 -- 通过
Python 调用：
import redis
class RedisFixedWindowLimiter:
def __init__(self, redis_client, limit, window_seconds):
self.r = redis_client
self.limit = limit
self.window = window_seconds
with open('rate_limit_fixed.lua') as f:
self.script = self.r.register_script(f.read())
def allow(self, key: str) -> bool:
result = self.script(keys=[key], args=[self.window, self.limit])
return result == 1
滑动窗口 Redis 实现（ZSET）
使用 Redis Sorted Set 记录每个请求的时间戳，精确统计窗口内请求数。
-- rate_limit_sliding.lua
-- KEYS[1]: 限流 key
-- ARGV[1]: 当前时间戳（毫秒）
-- ARGV[2]: 窗口大小（毫秒）
-- ARGV[3]: 限制次数
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])
local window_start = now - window
-- 清理过期记录
redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
-- 统计当前窗口内的请求数
local current = redis.call('ZCARD', key)
if current >= limit then
return 0
end
-- 记录当前请求
redis.call('ZADD', key, now, now .. ':' .. redis.call('INCR', 'req_counter'))
redis.call('PEXPIRE', key, window)
return 1
令牌桶 Redis 实现
-- rate_limit_token_bucket.lua
-- KEYS[1]: 令牌桶 key
-- ARGV[1]: 速率（每秒）
-- ARGV[2]: 容量
-- ARGV[3]: 当前时间（秒）
-- ARGV[4]: 需要的令牌数
local key = KEYS[1]
local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local needed = tonumber(ARGV[4])
local bucket = redis.call('HMGET', key, 'tokens', 'last_time')
local tokens = tonumber(bucket[1]) or capacity
local last_time = tonumber(bucket[2]) or now
-- 补充令牌
local delta = math.max(0, now - last_time)
tokens = math.min(capacity, tokens + delta * rate)
if tokens >= needed then
tokens = tokens - needed
redis.call('HMSET', key, 'tokens', tokens, 'last_time', now)
redis.call('EXPIRE', key, 60)
return 1
else
redis.call('HMSET', key, 'tokens', tokens, 'last_time', now)
redis.call('EXPIRE', key, 60)
return 0
end
面试答题框架
第一步：明确需求（30秒）
我需要确认：限流维度（用户/IP/API）、限流目标（保护下游还是公平使用）、是否需要分布式。
第二步：算法选择（1分钟）
我推荐令牌桶作为通用方案：
- 允许合理突发（用户体验好）
- 长期速率可控（保护服务）
- 实现简单，内存开销低
第三步：单机实现（2分钟）
展示 TokenBucket 的 Python 实现，讲解令牌补充和消耗的逻辑。
第四步：分布式扩展（2分钟）
使用 Redis + Lua 原子脚本，解决多机并发下的竞态条件，分别展示固定窗口、滑动窗口和令牌桶的 Redis 实现。
第五步：高级话题（2分钟，面试官追问时展开）
- 多级限流：网关层全局限流 + 服务层接口限流 + 用户维度精细限流
- 自适应限流：基于 CPU、延迟等指标动态调整阈值
- 限流后的降级策略：排队、拒绝、返回缓存、返回简化版数据
常见问题
Q：漏桶和令牌桶的核心区别？
漏桶是请求入桶、以固定速率出桶处理，满了就拒绝（类似队列）。令牌桶是请求拿令牌，有令牌就立即处理，无令牌才等待/拒绝。令牌桶更容易实现且允许突发。
Q：滑动窗口为什么用 Redis ZSET？
ZSET 的
ZREMRANGEBYSCORE可以高效删除过期记录，ZCARD快速统计数量，天然适合时间窗口统计。
Q：限流和熔断、降级的区别？
- 限流：控制请求速率，防止系统过载
- 熔断：当错误率过高时，快速失败，给系统恢复时间
- 降级：系统过载时，关闭非核心功能，保证核心功能可用
Q：Google 的 BBR 拥塞控制算法和限流有什么关系？
BBR（Bottleneck Bandwidth and RTT）通过实时测量网络带宽和延迟来动态调整发送速率，可以借鉴到自适应限流中：根据系统负载动态调整令牌桶的填充速率。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。