4. 滑动窗口算法详解：原理、参数、实现方式与优缺点

滑动窗口（Sliding Window）是限流领域中最常用的算法之一，它通过一个随时间移动的“窗口”来统计请求数量，从而在固定窗口的简单性和滑动日志的精确性之间取得了良好的平衡。本节将深入剖析其核心原理、关键参数、多种实现方式以及适用场景。

4.1 核心原理与数学模型

滑动窗口算法的核心思想是：将时间划分为多个细粒度的小格子（例如，将1秒划分为10个100ms的格子），每个格子记录该时间段内的请求计数。窗口随着时间“滑动”，每次滑动时，丢弃最旧的格子数据，加入最新的格子数据，并累加整个窗口内的请求总数。

与固定窗口算法相比，滑动窗口避免了“临界突变”问题。例如，固定窗口在1秒的边界处可能允许2倍于阈值的流量通过，而滑动窗口通过细粒度滑动，将流量平滑地分布在时间轴上。

4.2 关键参数详解

滑动窗口算法的行为由以下三个核心参数决定。合理配置这些参数是限流系统设计的关键。

4.3 实现方式

滑动窗口算法有多种实现方式，从简单的内存数据结构到基于Redis的分布式方案。以下是两种最常用的实现。

4.3.1 基于循环数组的内存实现

适用于单机限流场景。使用一个固定长度的数组（循环队列）来存储每个格子的计数，配合一个总计数器来避免每次滑动时重新求和。

class SlidingWindowCounter:
    def __init__(self, window_size_ms=1000, bucket_size_ms=200, threshold=100):
        self.window_size = window_size_ms
        self.bucket_size = bucket_size_ms
        self.threshold = threshold
        self.num_buckets = window_size_ms // bucket_size_ms
        # 循环数组，存储每个格子的计数
        self.buckets = [0] * self.num_buckets
        self.total_count = 0
        self.current_bucket_index = 0
        self.last_bucket_time = self._current_time_ms()

    def _current_time_ms(self):
        import time
        return int(time.time() * 1000)

    def allow_request(self):
        now = self._current_time_ms()
        # 计算需要滑动的格子数
        elapsed = now - self.last_bucket_time
        buckets_to_slide = elapsed // self.bucket_size
        if buckets_to_slide > 0:
            # 滑动窗口：丢弃旧格子，更新总计数
            for _ in range(min(buckets_to_slide, self.num_buckets)):
                self.current_bucket_index = (self.current_bucket_index + 1) % self.num_buckets
                self.total_count -= self.buckets[self.current_bucket_index]
                self.buckets[self.current_bucket_index] = 0
            self.last_bucket_time = now
        # 检查阈值
        if self.total_count < self.threshold:
            self.buckets[self.current_bucket_index] += 1
            self.total_count += 1
            return True
        return False

4.3.2 基于Redis Sorted Set的分布式实现

适用于分布式系统。利用Redis的ZSET（有序集合），将每个请求的时间戳作为score，通过ZREMRANGEBYSCORE移除窗口外的旧数据，再用ZCARD统计窗口内请求数。

import redis
import time

class RedisSlidingWindow:
    def __init__(self, redis_client, key, window_size=1, threshold=100):
        self.redis = redis_client
        self.key = key
        self.window_size = window_size  # 单位：秒
        self.threshold = threshold

    def allow_request(self):
        now = time.time()
        window_start = now - self.window_size
        pipe = self.redis.pipeline()
        # 1. 移除窗口外的旧记录
        pipe.zremrangebyscore(self.key, 0, window_start)
        # 2. 统计当前窗口内的请求数
        pipe.zcard(self.key)
        # 3. 添加当前请求（先添加再判断，或先判断再添加）
        pipe.zadd(self.key, {str(now): now})
        # 4. 设置过期时间，防止内存泄漏
        pipe.expire(self.key, self.window_size * 2)
        results = pipe.execute()
        current_count = results[1]  # zcard的结果
        if current_count > self.threshold:
            # 超过阈值，移除刚添加的请求
            self.redis.zrem(self.key, str(now))
            return False
        return True

4.4 优缺点分析

滑动窗口算法并非银弹，它在不同场景下表现出不同的优劣。下表对比了其核心优缺点。

4.5 适用场景与选型建议

滑动窗口算法是限流领域的“多面手”，但并非所有场景都适合。以下给出具体的选型建议：

• 推荐使用：
  • API网关的通用限流（如每秒1000次请求）。
  • 需要防止流量突刺但允许一定突发的中等精度场景。
  • 单机或小规模分布式系统（节点数 < 20）。
• 谨慎使用：
  • 对精度要求极高（如金融交易）—— 建议使用滑动日志（Sliding Log）算法。
  • 需要严格平滑输出（如视频流）—— 建议使用漏桶算法。
  • 超大规模分布式系统（节点数 > 100）—— Redis ZSET 方案可能成为瓶颈，建议考虑分布式令牌桶。