4. 滑动窗口算法实战：窗口大小与格子数设计、边界效应处理、基于Redis的滑动窗口实现思路

4.1 滑动窗口算法原理与固定窗口的缺陷

固定窗口算法（如每秒100次、每分钟1000次）存在明显的“突刺”问题：在窗口切换的瞬间，流量可以翻倍。例如，一个1分钟窗口允许1000次请求，若第59秒和第60秒各通过1000次，则实际2秒内通过了2000次请求，远超阈值。

滑动窗口算法通过将时间窗口细分为多个小格子（子窗口），并随当前时间不断向前滑动，从而平滑流量控制，消除边界突刺。

4.2 窗口大小与格子数设计原则

窗口大小（W）和格子数（N）直接影响限流的精度与性能。格子数越多，精度越高，但存储和计算开销也越大。

4.3 边界效应处理

滑动窗口的边界效应主要体现在以下三个方面，处理不当会导致限流失效或误判：

• 格子边界对齐： 请求时间戳必须对齐到格子起始时间。例如格子时长100ms，时间戳1630000000123ms应对齐到1630000000100ms（向下取整到100ms倍数）。
• 窗口滑动滞后： 当前时间位于格子中间时，窗口包含当前格子+前(N-1)个完整格子。不能将未完成的格子视为完整格子。
• 并发写入冲突： 同一格子可能被多个请求同时写入，需使用原子操作（如Redis INCR）或分布式锁。

4.4 基于Redis的滑动窗口实现思路

Redis是实现分布式滑动窗口限流的首选中间件。核心思路是使用有序集合（Sorted Set）或哈希表（Hash）存储每个格子的计数器。

方案一：基于Sorted Set（精确计数）

每个请求的时间戳作为score，成员为唯一标识（如UUID）。通过 ZREMRANGEBYSCORE 移除窗口外的旧数据，ZCARD 获取当前窗口请求数。

# 伪代码：基于Sorted Set的滑动窗口
def sliding_window_ss(user_id, limit=100, window_ms=1000):
    key = f"sliding:{user_id}"
    now = int(time.time() * 1000)
    window_start = now - window_ms

    # 移除窗口外的旧数据
    redis.zremrangebyscore(key, 0, window_start)

    # 获取当前窗口请求数
    current = redis.zcard(key)

    if current >= limit:
        return False  # 限流

    # 添加当前请求
    redis.zadd(key, {str(uuid.uuid4()): now})
    redis.expire(key, window_ms // 1000 + 1)  # 设置过期时间
    return True

方案二：基于Hash + 格子计数器（推荐，性能更优）

将时间窗口划分为N个格子，每个格子对应一个Hash field，使用 HINCRBY 原子递增。计算当前窗口总和时，只需读取N个field并求和。

# 伪代码：基于Hash的滑动窗口（格子数N=10）
def sliding_window_hash(user_id, limit=100, window_ms=1000, slots=10):
    key = f"sliding_hash:{user_id}"
    slot_ms = window_ms // slots
    now = int(time.time() * 1000)
    slot_id = (now // slot_ms) % slots  # 当前格子索引
    window_start_slot = (now - window_ms) // slot_ms

    # 原子递增当前格子
    redis.hincrby(key, slot_id, 1)

    # 计算窗口内所有格子总和
    total = 0
    for i in range(slots):
        # 计算格子对应的slot_id（考虑循环）
        slot = (slot_id - i) % slots
        # 判断该格子是否在窗口内
        slot_time = (now // slot_ms) * slot_ms - i * slot_ms
        if slot_time >= now - window_ms:
            total += int(redis.hget(key, slot) or 0)

    # 设置过期时间
    redis.expire(key, window_ms // 1000 * 2)

    return total <= limit

4.5 格子数选择与精度权衡

格子数N的选择需要在精度与性能之间取得平衡。以下表格展示了不同N值下的效果：

4.6 生产环境注意事项

• Redis key 过期策略： 设置TTL为窗口时长的2倍，避免key长期占用内存。例如W=60s，TTL=120s。
• 时钟同步： 所有应用服务器时间需通过NTP同步，否则格子对齐会错乱。误差应小于格子时长的10%。
• 原子性与Lua脚本： 将“递增+求和+判断”封装为Lua脚本，在Redis端原子执行，避免竞态条件。
• 降级方案： 当Redis不可用时，应降级为本地内存滑动窗口（单机限流），或直接放行（需评估风险）。

-- Lua脚本：原子滑动窗口限流（Hash方案）
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window_ms = tonumber(ARGV[2])
local slots = tonumber(ARGV[3])
local now = tonumber(ARGV[4])

local slot_ms = window_ms / slots
local slot_id = math.floor(now / slot_ms) % slots
local window_start = now - window_ms

-- 递增当前格子
redis.call('HINCRBY', key, slot_id, 1)

-- 计算窗口内总和
local total = 0
for i = 0, slots - 1 do
    local s = (slot_id - i) % slots
    local slot_time = math.floor(now / slot_ms) * slot_ms - i * slot_ms
    if slot_time >= window_start then
        local count = redis.call('HGET', key, s)
        if count then
            total = total + tonumber(count)
        end
    end
end

-- 设置过期时间
redis.call('EXPIRE', key, window_ms / 1000 * 2)

if total > limit then
    return 0  -- 限流
else
    return 1  -- 放行
end