3. 滑动窗口算法：原理详解、Python实现与精度优化

固定窗口算法在窗口切换瞬间存在流量突刺的漏洞，而滑动窗口算法通过将时间窗口细分为多个更小的“子窗口”（Slot），以更精细的粒度统计请求量，从而平滑流量控制，是金融级限流场景中的核心算法之一。

3.1 滑动窗口核心原理

滑动窗口不再使用单一的全局计数器，而是将一个完整的时间窗口（例如 1 秒）划分为 N 个等长的子窗口。每个子窗口维护独立的计数器。当请求到达时，算法计算当前时间点所属的窗口范围（包含当前子窗口及之前的 N-1 个子窗口），累加这些子窗口的计数，若总和超过阈值则触发限流。

对比维度	固定窗口	滑动窗口
统计粒度	整个窗口一个计数器	窗口内多个子窗口计数器
边界突刺	窗口切换时允许 2 倍阈值流量	平滑过渡，无突刺
内存占用	O(1)	O(N)，N 为子窗口数量
精度控制	粗粒度	细粒度，精度由 N 决定
适用场景	非关键路径、允许毛刺	金融交易、API 网关核心限流

3.2 Python 基础实现

以下实现使用 collections.deque 存储子窗口记录，每个元素为 (timestamp, count) 元组。核心逻辑包括：清理过期子窗口、累加有效计数、判断是否限流。

import time
from collections import deque

class SlidingWindowLimiter:
    """滑动窗口限流器（基础版）"""
    def __init__(self, max_requests: int, window_size: float = 1.0, slots: int = 10):
        """
        :param max_requests: 窗口内允许的最大请求数
        :param window_size: 窗口大小（秒）
        :param slots: 子窗口数量（精度控制）
        """
        self.max_requests = max_requests
        self.window_size = window_size
        self.slot_duration = window_size / slots  # 每个子窗口时长
        self.slots = deque()  # 存储 (slot_start_time, count)

    def _get_current_slot(self) -> float:
        """计算当前时间所属的子窗口起始时间"""
        now = time.time()
        return now - (now % self.slot_duration)

    def _clean_expired_slots(self):
        """移除所有已过期的子窗口"""
        now = time.time()
        window_start = now - self.window_size
        while self.slots and self.slots[0][0] < window_start:
            self.slots.popleft()

    def allow_request(self) -> bool:
        """判断是否允许当前请求"""
        self._clean_expired_slots()
        current_slot_start = self._get_current_slot()

        # 更新或追加当前子窗口
        if self.slots and self.slots[-1][0] == current_slot_start:
            # 当前子窗口已存在，增加计数
            self.slots[-1] = (current_slot_start, self.slots[-1][1] + 1)
        else:
            # 创建新的子窗口
            self.slots.append((current_slot_start, 1))

        # 计算当前滑动窗口内总请求数
        total_requests = sum(count for _, count in self.slots)
        return total_requests <= self.max_requests

3.3 精度优化：子窗口数量与内存权衡

滑动窗口的精度由子窗口数量 slots 决定。slots 越大，统计粒度越细，但内存占用和计算开销也线性增长。金融场景通常需要在精度与性能之间找到平衡点。

3.3.1 精度与性能对比

子窗口数 (slots)	理论最大误差	内存占用 (deque)	推荐场景
5	20% 窗口大小	~5 个元素	非关键路径、日志采样
10	10% 窗口大小	~10 个元素	通用 API 限流
20	5% 窗口大小	~20 个元素	金融交易、高精度控制
50	2% 窗口大小	~50 个元素	高频交易、毫秒级限流

3.3.2 优化策略：动态子窗口合并

在高并发场景下，固定 slots 可能导致频繁的 deque 操作。优化思路是：当子窗口内无请求时，跳过创建；当连续多个子窗口计数为 0 时，合并为一个过期标记，减少元素数量。

class OptimizedSlidingWindowLimiter(SlidingWindowLimiter):
    """优化版：跳过空子窗口，减少内存碎片"""
    def allow_request(self) -> bool:
        self._clean_expired_slots()
        current_slot_start = self._get_current_slot()

        # 优化：如果当前子窗口与最后一个子窗口不连续，插入空窗口标记
        if self.slots and self.slots[-1][0] < current_slot_start - self.slot_duration:
            # 存在时间间隙，插入一个空窗口（仅用于占位，不增加计数）
            self.slots.append((current_slot_start - self.slot_duration, 0))

        # 更新或追加当前子窗口
        if self.slots and self.slots[-1][0] == current_slot_start:
            self.slots[-1] = (current_slot_start, self.slots[-1][1] + 1)
        else:
            self.slots.append((current_slot_start, 1))

        total_requests = sum(count for _, count in self.slots)
        return total_requests <= self.max_requests

3.4 实战案例：金融 API 限流配置

假设某金融数据 API 的限流规则为：每秒最多 100 次请求，要求误差不超过 10%。根据精度要求，选择 slots=10，每个子窗口 100ms，理论最大误差 10%。

# 实战配置示例
limiter = SlidingWindowLimiter(
    max_requests=100,
    window_size=1.0,
    slots=10  # 每个子窗口 100ms，误差 ≤ 10%
)

# 模拟请求
for i in range(120):
    if limiter.allow_request():
        print(f"请求 {i+1}: 允许")
    else:
        print(f"请求 {i+1}: 限流（拒绝）")
    time.sleep(0.005)  # 模拟 5ms 间隔

3.5 滑动窗口 vs 其他算法选型建议

算法	精度	内存	实现复杂度	金融推荐度
固定窗口	低	低	极低	★☆☆☆☆
滑动窗口	中-高	中	中	★★★★☆
令牌桶	高	低	中	★★★★★
漏桶	高	低	低-中	★★★☆☆