Redis 限流不完全指南：为什么你的计数器会被并发打爆？

做后端开发的，谁没遇到过接口被刷？活动一上线，流量比预期多三个零，数据库直接原地升天。这时候老板会问你："加个限流不行吗？"

行，太行了。但你真的了解限流吗？

今天来聊一个看起来很简单，实际上到处都是坑的话题：基于 Redis 的分布式限流。从最土的计数器到稍微优雅一点的令牌桶，手把手带你避坑。

第一层：最 naive 的方案——滑动窗口计数器

大多数人的第一反应是这样的：

# Lua 脚本
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local window_start = now - window

-- 删除窗口外的数据
redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
-- 统计窗口内请求数
local count = redis.call('ZCARD', key)

if count >= limit then
    return 0  -- 被限流
end

-- 记录本次请求
redis.call('ZADD', key, now, now)
redis.call('EXPIRE', key, window)
return 1

看着没问题对吧？逻辑清晰，代码简洁。

但是！你用 ZSET 做滑动窗口，每次请求都要 ZREMRANGEBYSCORE + ZCARD + ZADD + EXPIRE，四次网络往返。高并发下光 Redis 网络开销就能让你哭。

更骚的是，如果你的服务部署在多个节点，这个方案本身没问题——但如果你在网关层做限流，多个网关节点的 ZSET 是共享的（因为连同一个 Redis），这时候就会出现大量 ZREMRANGEBYSCORE 操作竞争。

教训：滑动窗口精度高，但代价也高。除非你真的需要精确的时间窗口边界，否则别用它。

第二层：固定窗口计数器——看起来简单，用起来全是雷

有人说了，ZSET 太重了，我用简单的 INCR 不就完了？

local key = KEYS[1]
local limit = tonumber(ARGV[1])
local expire = tonumber(ARGV[2])

local count = redis.call('INCR', key)
if count == 1 then
    redis.call('EXPIRE', key, expire)
end

if count > limit then
    return 0
end
return 1

代码行数少了一大半，性能也好。但是——固定窗口边界问题了解一下？

假设你设置每秒最多 100 请求。你的窗口是 [t, t+1s)。用户在 t+0.99s 发了 100 个请求，全部通过；然后在 t+1.01s 又发了 100 个请求，因为新的窗口开始了，又全部通过。

实际效果：0.2 秒内处理了 200 个请求，瞬时 QPS 是限制值的 5 倍。

这个问题在流量突增场景下尤为明显。你的限流形同虚设。

第三层：令牌桶——终于有个像样的方案了

令牌桶的核心思想很简单：有一个桶，桶里有 N 个令牌。每来一个请求，必须从桶里取一个令牌，取到才能过；桶里的令牌以固定速率补充。

用 Redis 怎么实现？有一种经典写法叫"预消费"模式：

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 last_refill = tonumber(redis.call('HGET', key, 'last_refill')) or now
local tokens = tonumber(redis.call('HGET', key, 'tokens')) or capacity

-- 计算应该补充多少令牌
local elapsed = now - last_refill
local filled = math.min(capacity, tokens + (elapsed * rate))

if filled >= requested then
    -- 取走令牌
    redis.call('HSET', key, 'last_refill', now)
    redis.call('HSET', key, 'tokens', filled - requested)
    return 1  -- 允许通过
end
return 0  -- 拒绝

这里有个致命 bug：当 Redis 主从切换时，HGET 可能读到过期数据，导致令牌数计算错误。更要命的是，如果多个请求并发执行，Redis 单线程执行 Lua 脚本天然保证了原子性——但如果你用的是 Redis Cluster，哈希槽分散到不同节点，这个 key 就没法保证了。

解决方案：用 tag 属性强制让这个 key 的所有字段落到同一个 slot：

local key = '{rate_limit}:' .. KEYS[1]

加个大括号，让 Redis 把整个 key 当成一个整体处理。

第四层：漏桶算法——和令牌桶相爱相杀

很多人搞不清令牌桶和漏桶的区别。

令牌桶：桶里有令牌才能取，突发流量可以被消化（一次性取多个令牌）。
漏桶：以固定速率漏出，水龙头开多大都没用，突发流量会被强制平滑。

-- 漏桶实现：简单粗暴
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 water = tonumber(redis.call('GET', key)) or 0
local new_water = math.max(0, water - ((now - (tonumber(redis.call('GET', key .. ':last') or now))) * rate))

if new_water + requested <= capacity then
    redis.call('SET', key, new_water + requested)
    redis.call('SET', key .. ':last', now)
    return 1
end
return 0

漏桶的问题是：无法应对突发流量。你期望系统在正常情况下能扛住正常流量，突发时快速响应，但漏桶会强制你排队。

所以结论是：大多数 API 限流场景用令牌桶，除非你有特殊的平滑需求。

实战：结合 Lua + Redis 实现一个生产级限流中间件

光有算法不够，来点实际能跑的。这里用 Python + Redis 演示一个基于令牌桶的装饰器：

import redis
import time
import functools
from typing import Tuple

class RateLimiter:
    def __init__(self, redis_client: redis.Redis, key_prefix: str = "rl"):
        self.redis = redis_client
        self.key_prefix = key_prefix
    
    def _lua_script(self) -> str:
        return """
        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 last_refill = tonumber(redis.call('HGET', key, 'last_refill')) or now
        local tokens = tonumber(redis.call('HGET', key, 'tokens')) or capacity

        local elapsed = now - last_refill
        local filled = math.min(capacity, tokens + (elapsed * rate))

        if filled >= requested then
            redis.call('HSET', key, 'last_refill', now)
            redis.call('HSET', key, 'tokens', filled - requested)
            return 1
        end
        return 0
        """

    def is_allowed(self, key: str, rate: int, capacity: int, requested: int = 1) -> bool:
        full_key = f"{{{self.key_prefix}}}:{key}"
        script = self._lua_script()
        result = self.redis.eval(
            script, 1, full_key, rate, capacity, time.time(), requested
        )
        return result == 1

# 使用方式
limiter = RateLimiter(redis.Redis(host='localhost', port=6379))

def rate_limit(rate: int, capacity: int):
    """装饰器：每秒rate个请求，桶容量capacity"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = f"func:{func.__name__}"
            if not limiter.is_allowed(key, rate, capacity):
                raise Exception("请求太频繁，请稍后重试")
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit(rate=10, capacity=20)
def heavy_api():
    return "OK"

这个方案的好处：

• 原子操作：Lua 脚本在 Redis 端执行，不存在并发问题
• 单机精度：单个 Redis 实例下精度完全有保障
• 可扩展：改变 rate 和 capacity 参数即可适配不同接口

分布式场景的终极问题：Redis 挂了怎么办？

很多人忽略了这一点：你的限流依赖 Redis，如果 Redis 挂了，限流逻辑怎么走？

两种策略：

1. failopen（默认）：Redis 挂了就不限流，直接放行。好处是不影响业务，坏处是 Redis 挂了的那段时间接口完全暴露。适用于限流不是核心兜底手段的场景。
2. failclosed：Redis 挂了默认拒绝。好处是安全，坏处是可能误杀正常流量。适用于限流是核心兜底手段的场景（如支付接口、防爬）。

生产环境建议：failopen + 监控告警。你需要知道 Redis 什么时候挂了。

def is_allowed_safe(self, key: str, rate: int, capacity: int) -> bool:
    try:
        return self.is_allowed(key, rate, capacity)
    except redis.RedisError:
        # Redis 挂了，走 failopen 策略并告警
        logger.error(f"Rate limiter Redis error, allowing request: {key}")
        return True

总结：限流方案选型指南

聊了这么多，来个总结：

• 固定窗口计数器：性能好，但精度差，有窗口边界问题。适合对精度要求不高的场景。
• 滑动窗口：精度高，但实现复杂，性能开销大。适合需要精确时间边界的场景。
• 令牌桶：精度和性能兼顾，支持突发流量。适合大多数 API 限流场景。首选。
• 漏桶：强制平滑流量，不支持突发。适合需要严格流量整形的场景（如消息队列限速）。

最后说一句：限流是防御性手段，不是性能优化手段。很多人把限流当成解决系统性能问题的万金油，这是方向性错误。如果你的系统被刷是因为性能不够，加多少限流都救不了你。

先把系统撑到不需要限流的程度，再用限流来兜底。这是正确的顺序。

好了，去写你的限流代码吧，别再被并发打爆了。