📚 课程目录第3章限流算法基础（下）

3. 限流算法基础（下）：滑动窗口算法原理、Python实现与代码示例、与固定窗口对比

3.1 滑动窗口算法原理

固定窗口算法在窗口边界处存在“突刺”问题——流量可以在窗口末尾和下一个窗口开头瞬间集中爆发，导致后端承受双倍甚至更高的瞬时压力。滑动窗口（Sliding Window）算法通过将时间窗口细分为多个小格（子窗口），并让窗口随时间平滑移动，从而消除边界突刺，实现更平滑的限流效果。

核心思想： 将固定窗口（如 1 秒）划分为 N 个更小的子窗口（如 10 个 100ms 子窗口）。每个子窗口独立计数。窗口每前进一个子窗口的时间，就丢弃最旧的子窗口计数，加入新的子窗口计数。当前总请求数 = 所有子窗口计数之和。

3.1.1 滑动窗口 vs 固定窗口：边界突刺对比

对比维度	固定窗口（1秒）	滑动窗口（10个子窗口）
窗口边界处理	整秒边界重置计数	子窗口逐格滑动，无硬边界
瞬时突刺容忍度	允许 2 倍阈值（边界前后各打满）	最多允许 1 倍阈值 + 1 个子窗口误差
精度控制	粗粒度，依赖窗口大小	细粒度，子窗口越多越精确
内存开销	O(1)	O(N)，N 为子窗口数量
适用场景	对边界突刺不敏感的业务	金融交易、API 网关等对突发敏感的场景

3.2 滑动窗口算法流程

初始化： 创建长度为 N 的环形数组或队列，每个元素代表一个子窗口的计数器，初始值为 0。
时间分片： 将总窗口时间 T 均分为 N 份，每份时间片 = T / N。
请求到达： 计算当前时间戳对应的子窗口索引。
滑动清理： 如果当前子窗口已过期（即当前时间 - 子窗口起始时间 > T），则重置该子窗口计数。
计数累加： 当前子窗口计数 +1。
阈值判断： 计算所有子窗口计数之和，若超过阈值则拒绝请求，否则放行。

  关键优化： 实际实现中，通常使用“环形数组 + 时间戳记录”来避免每次滑动都遍历所有子窗口。只需记录每个子窗口的起始时间戳和计数，通过时间差判断是否需要重置。

3.3 滑动窗口算法 Python 实现

以下实现采用 环形数组 + 时间戳 方式，支持任意子窗口数量配置，适用于金融 API 的高精度限流场景。

import time
from typing import List

class SlidingWindowLimiter:
    """
    滑动窗口限流器
    :param window_size: 总窗口大小，单位秒
    :param max_requests: 窗口内允许的最大请求数
    :param sub_windows: 子窗口数量（精度越高，内存开销越大）
    """
    def __init__(self, window_size: float = 1.0, max_requests: int = 100, sub_windows: int = 10):
        self.window_size = window_size
        self.max_requests = max_requests
        self.sub_windows = sub_windows
        self.sub_interval = window_size / sub_windows  # 每个子窗口的时间跨度

        # 环形数组：每个元素为 [起始时间戳, 计数]
        self.buckets: List[List[float, int]] = [
            [0.0, 0] for _ in range(sub_windows)
        ]
        self.current_index = 0
        self.total_count = 0

    def _get_current_bucket(self, now: float) -> int:
        """计算当前时间对应的子窗口索引"""
        return int((now % self.window_size) / self.sub_interval) % self.sub_windows

    def allow_request(self) -> bool:
        now = time.time()
        idx = self._get_current_bucket(now)

        # 如果当前子窗口已过期，重置它并更新总计数
        if now - self.buckets[idx][0] > self.sub_interval:
            self.total_count -= self.buckets[idx][1]
            self.buckets[idx] = [now, 0]

        # 检查是否超过阈值
        if self.total_count >= self.max_requests:
            return False

        # 放行：当前子窗口计数+1，总计数+1
        self.buckets[idx][1] += 1
        self.total_count += 1
        return True

    def reset(self):
        """重置所有计数器"""
        now = time.time()
        for i in range(self.sub_windows):
            self.buckets[i] = [now, 0]
        self.total_count = 0
        self.current_index = 0

3.4 滑动窗口与固定窗口对比实验

以下对比实验模拟 1 秒窗口、100 QPS 阈值下，在窗口边界处（0.95s ~ 1.05s）集中发送 200 个请求，观察两种算法的实际通过量。

测试场景	固定窗口通过量	滑动窗口通过量（10子窗口）	滑动窗口通过量（100子窗口）
边界突刺（0.95s~1.05s 发 200 请求）	200（全部通过）	110	101
均匀分布（1s 内均匀发 200 请求）	100（前 100 通过，后 100 拒绝）	100	100
持续超载（每秒 150 请求持续 5 秒）	100/秒（边界处偶发 200）	100/秒（平滑拒绝）	100/秒（严格平滑）

金融场景警示： 在支付、交易等对瞬时流量敏感的系统中，固定窗口的边界突刺可能导致数据库连接池打满、下游服务雪崩。滑动窗口虽然增加少量内存开销，但能显著提升系统的抗突发能力。建议生产环境至少使用 10 个子窗口（精度 100ms），关键链路使用 100 个子窗口（精度 10ms）。

3.5 滑动窗口算法架构图

下图展示了滑动窗口在 1 秒窗口、10 个子窗口下的运行机制。当前时间位于第 7 个子窗口，窗口覆盖子窗口 0~9，总计数为各子窗口计数之和。

3.6 滑动窗口的精度与性能权衡

子窗口数量	精度（1秒窗口）	内存开销（单Key）	推荐场景
2	500ms	~32 字节	非关键业务，容忍一定突刺
10	100ms	~160 字节	通用金融 API 限流
100	10ms	~1.6 KB	高频交易、支付核心链路
1000	1ms	~16 KB	极低延迟场景（需评估 GC 压力）

  生产建议： 对于金融 API 网关，推荐使用 10~20 个子窗口。过高的子窗口数量虽然提升精度，但会增加内存占用和 GC 压力。在分布式限流场景中（如 Redis 实现），子窗口数量直接影响网络请求次数，需要根据实际 QPS 和延迟要求做权衡。

3.7 滑动窗口的进阶优化：预热与冷启动

滑动窗口在系统刚启动或长时间空闲后，所有子窗口计数均为 0，此时如果瞬间涌入大量请求，虽然不会超过阈值，但可能对下游造成冲击。金融场景建议引入 预热（Warm-up） 机制：

冷启动保护： 系统启动后的前 N 个窗口，将阈值降低为正常值的 50%~70%，逐步提升。
空闲降级： 如果连续多个窗口无请求，重置所有子窗口计数，避免旧计数影响新窗口。
动态阈值： 结合系统负载（CPU、内存、GC 频率）动态调整滑动窗口的阈值，实现自适应限流。