Redis限流算法：从固定窗口到滑动窗口与令牌桶的演进

2026/10/31 3:25:17

1. 固定窗口计数与INCR命令的局限性

固定窗口计数是限流算法中最基础的一种实现方式，它的核心思想是在固定的时间窗口内（比如1分钟）统计请求次数，当超过阈值时就拒绝后续请求。Redis的INCR命令看似是实现这个算法的完美选择，但实际上存在几个关键缺陷。

1.1 时间边界问题

使用INCR实现固定窗口最直接的方式是为每个时间窗口创建一个独立的计数器。例如对于每分钟限流100次的场景：

# 伪代码示例 current_minute = get_current_minute() key = "rate_limit:" + service_name + ":" + current_minute count = INCR(key) EXPIRE(key, 60) if count > 100 then reject_request()

这种实现存在明显的时间边界问题。假设在时间窗口切换的瞬间（比如10:00:59和10:01:00之间），系统可能在前一窗口的最后1秒和后一窗口的第1秒各接收100个请求，实际上在2秒内处理了200个请求，这完全违背了限流的初衷。

1.2 内存膨胀风险

在高并发系统中，如果为每个时间窗口都创建独立的key，会导致Redis内存中堆积大量短期使用的key。特别是当限流维度较多时（比如按用户ID+接口维度限流），内存占用会呈指数级增长。

我曾经在一个生产环境中见过这样的案例：一个按用户维度限流的系统，在用户量突增时，Redis内存使用量在几分钟内从2GB飙升到16GB，最终导致OOM崩溃。

2. 更优的替代方案

2.1 滑动窗口算法

滑动窗口算法通过记录更细粒度的时间戳来解决边界问题。典型实现方式：

# 使用ZSET存储请求时间戳 current_time = get_current_timestamp() window_start = current_time - 60_000 # 60秒窗口 ZREMRANGEBYSCORE(key, 0, window_start) # 移除旧时间戳 ZADD(key, current_time, current_time) # 添加当前时间戳 count = ZCARD(key) # 获取当前窗口内请求数 EXPIRE(key, 120) # 设置稍长的过期时间 if count > 100 then reject_request()

这种实现虽然更精确，但代价是更高的内存消耗和计算开销。每个请求都需要存储完整的时间戳，且需要频繁执行ZREMRANGEBYSCORE操作。

2.2 令牌桶算法

令牌桶算法通过恒定速率生成令牌的方式实现更平滑的限流：

# 使用Hash存储令牌数和最后更新时间 current_time = get_current_timestamp() data = HGETALL(key) tokens = data.tokens || 100 last_time = data.last_time || current_time # 计算新增令牌数 elapsed = current_time - last_time new_tokens = elapsed * (rate_per_second) tokens = min(tokens + new_tokens, capacity) if tokens >= 1 then tokens -= 1 HSET(key, "tokens", tokens, "last_time", current_time) EXPIRE(key, 120) process_request() else reject_request()

令牌桶算法能更好地应对突发流量，允许短时间内超过平均速率，同时保证长期平均速率不超过阈值。

3. 生产环境实践建议

3.1 分布式限流的一致性

在分布式系统中实现精确限流需要考虑原子性问题。Redis的Lua脚本是很好的解决方案：

local key = KEYS[1] local now = tonumber(ARGV[1]) local window = tonumber(ARGV[2]) local limit = tonumber(ARGV[3]) local clearBefore = now - window redis.call('ZREMRANGEBYSCORE', key, 0, clearBefore) redis.call('ZADD', key, now, now) redis.call('EXPIRE', key, window/1000 * 2) local count = redis.call('ZCARD', key) if count > limit then return 0 else return 1 end

3.2 多级限流策略

在实际系统中，建议采用多级限流策略：

全局基础限流（最宽松）
服务/接口级限流
用户/业务维度限流（最严格）

这种分层结构既能保护系统整体稳定性，又能保证重要业务的可用性。

4. 性能优化技巧

4.1 内存优化

对于滑动窗口实现，可以采用以下优化：

使用微秒时间戳而非毫秒，减少存储空间
定期清理过期数据，避免ZSET过大
对相同时间戳的请求做合并计数

4.2 计算优化

在令牌桶算法中，可以预计算下一次可用时间，避免频繁查询Redis：

def check_rate_limit(user_id): now = time.time() next_available = cache.get(f"next_available:{user_id}") if next_available and now < next_available: return False # 真正的Redis操作 allowed = redis_ratelimit(user_id) if not allowed: cache.set(f"next_available:{user_id}", now + 1, ex=1) return allowed

这种本地缓存+预计算的模式可以减少80%以上的Redis查询。