Redis 计数器与限流

Redis的INCR等原子命令完美支持计数器，结合过期和Lua脚本实现多种限流策略。

计数器实现

简单计数器

Bash

# 初始化计数器
SET counter:page:views 0

# 自增
INCR counter:page:views
# 返回: 1

# 自增指定值
INCRBY counter:page:views 100

# 自减
DECR counter:page:views

# 获取当前值
GET counter:page:views

计数器特点

Bash

- INCR/DECR原子操作
- 不存在时自动从0开始
- 值为整数，范围：-2^63 到 2^63-1

分布式计数器

多节点计数

Bash

# 各节点本地计数
INCR counter:node1:views

# 定时汇总
total = node1 + node2 + node3

# 或直接全局计数
INCR counter:global:views

Hash计数

Bash

# 多维度计数
HINCRBY stats:page:home views 1
HINCRBY stats:page:home unique_visitors 1
HINCRBY stats:page:home clicks 5

# 获取全部统计
HGETALL stats:page:home

时间窗口计数

每日计数

Bash

# 每日访问计数
INCR counter:daily:2024:01:01

# 设置次日过期
EXPIRE counter:daily:2024:01:01 86400 * 2

# 获取今日计数
GET counter:daily:2024:01:01

每小时计数

Bash

# 小时计数
INCR counter:hourly:2024:01:01:10

# 设置过期
EXPIRE counter:hourly:2024:01:01:10 3600 * 24

滑动时间窗口

Python

# 使用ZSet实现滑动窗口
ZADD window:user:1000 1700001000 "req1"
ZADD window:user:1000 1700002000 "req2"

# 移除过期请求
ZREMRANGEBYSCORE window:user:1000 0 {current_time - window_size}

# 统计窗口内请求数
ZCARD window:user:1000

固定窗口限流

实现原理

Python

固定时间窗口：
- 每分钟/每小时为一个窗口
- 突发流量可能绕过限制（边界问题）
- 实现简单

固定窗口实现

Python

def fixed_window_rate_limit(key, limit, window_seconds):
    current_window = int(time.time() / window_seconds)
    counter_key = f"{key}:{current_window}"

    # 获取当前窗口计数
    count = redis.incr(counter_key)

    # 首次计数设置过期
    if count == 1:
        redis.expire(counter_key, window_seconds)

    # 判断是否超限
    if count > limit:
        return False  # 限流

    return True  # 通过

固定窗口缺点

Python

边界问题：
- 11:00:59突发100请求
- 12:00:01又突发100请求
- 短时间内200请求绕过100的限制

滑动窗口限流

ZSet实现滑动窗口

lua

def sliding_window_rate_limit(key, limit, window_seconds):
    now = time.time()
    window_start = now - window_seconds

    # 移除窗口外的请求
    redis.zremrangebyscore(key, 0, window_start)

    # 统计窗口内请求数
    count = redis.zcard(key)

    if count >= limit:
        return False  # 限流

    # 添加当前请求
    redis.zadd(key, {str(now): now})

    # 设置过期时间
    redis.expire(key, window_seconds)

    return True  # 通过

滑动窗口特点

Python

- 窗口平滑滑动
- 解决固定窗口边界问题
- 精确控制时间窗口
- 内存占用较高（存储每次请求时间戳）

令牌桶限流

令牌桶原理

Python

- 以固定速率生成令牌放入桶中
- 桶有最大容量
- 请求消耗一个令牌
- 令牌不足则拒绝
- 允许突发流量（桶内令牌积累）

令牌桶实现

Python

def token_bucket_rate_limit(key, rate, capacity):
    now = time.time()

    # 获取上次更新时间和当前令牌数
    last_time = redis.hget(key, "last_time") or 0
    tokens = redis.hget(key, "tokens") or capacity

    # 计算新令牌数
    elapsed = now - float(last_time)
    new_tokens = min(capacity, float(tokens) + elapsed * rate)

    if new_tokens >= 1:
        # 消耗一个令牌
        redis.hset(key, "tokens", new_tokens - 1)
        redis.hset(key, "last_time", now)
        redis.expire(key, int(capacity / rate) + 10)
        return True

    return False  # 令牌不足，限流

漏桶限流

漏桶原理

Python

- 请求以任意速率进入桶
- 桶以固定速率漏水（处理请求）
- 桶满则拒绝新请求
- 输出流量平滑
- 不允许突发流量

漏桶实现

text

def leaky_bucket_rate_limit(key, rate, capacity):
    now = time.time()

    # 获取上次漏水时间和桶内水量
    last_time = redis.hget(key, "last_time") or now
    water = redis.hget(key, "water") or 0

    # 计算漏水量
    elapsed = now - float(last_time)
    leaked = elapsed * rate
    water = max(0, float(water) - leaked)

    # 检查桶是否溢出
    if water < capacity:
        # 加入一滴水
        redis.hset(key, "water", water + 1)
        redis.hset(key, "last_time", now)
        redis.expire(key, int(capacity / rate) + 10)
        return True

    return False  # 桶满，限流

Lua脚本限流

滑动窗口Lua脚本

text

-- 滑动窗口限流脚本
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

-- 移除过期请求
redis.call('ZREMRANGEBYSCORE', key, 0, now - window)

-- 统计当前数量
local count = redis.call('ZCARD', key)

if count < limit then
    -- 添加请求
    redis.call('ZADD', key, now, now)
    redis.call('EXPIRE', key, window)
    return 1  -- 通过
else
    return 0  -- 限流
end

使用脚本

text

script = "
...Lua脚本内容...
"
sha = redis.script_load(script)

def rate_limit_lua(key, limit, window):
    now = int(time.time() * 1000)
    return redis.evalsha(sha, 1, key, limit, window, now)

限流算法对比

算法	特点	突发流量	实现复杂度	适用场景
固定窗口	简单	允许边界突发	低	简单限流
滑动窗口	精确	严格控制	中	API限流
令牌桶	允许突发	允许	中	流量控制
漏桶	平滑输出	不允许	中	速率限制

应用场景

API限流

text

# 用户API调用限制
def api_rate_limit(user_id):
    key = f"rate_limit:api:{user_id}"
    return sliding_window_rate_limit(key, 100, 60)  # 每分钟100次

IP限流

text

# IP请求限制
def ip_rate_limit(ip):
    key = f"rate_limit:ip:{ip}"
    return fixed_window_rate_limit(key, 1000, 60)  # 每分钟1000次

接口防护

text

# 关键接口严格限流
def sensitive_api_limit(user_id):
    key = f"rate_limit:sensitive:{user_id}"
    return sliding_window_rate_limit(key, 10, 60)  # 每分钟10次

要点总结

INCR/DECR原子操作实现简单计数器
Hash实现多维度计数，HINCRBY原子增减
固定窗口：简单但有边界突发问题
滑动窗口：精确控制，ZSet实现
令牌桶：允许突发流量，适合流量控制
漏桶：平滑输出，适合速率限制
Lua脚本保证限流逻辑原子性
根据场景选择限流算法：API限流用滑动窗口，流量控制用令牌桶
设置合理限流阈值和窗口时间

📝 发现内容有误？点击此处直接编辑