4、滑动窗口算法详解：滑动日志与滑动窗口、内存优化方案、代码实战（Python实现）

限流算法里，固定窗口有个硬伤——窗口切换瞬间的流量尖刺。你想想看，如果用户在 09:59:59 发了 100 个请求，又在 10:00:00 发了 100 个请求，固定窗口会放行 200 个。这显然不合理。

滑动窗口就是为了解决这个问题。说白了，它把时间窗口切成更小的格子，让限流判断更平滑。我在项目中见过不少团队因为固定窗口的毛刺问题导致线上告警，最后都换成了滑动窗口。

4.1 滑动日志 vs 滑动窗口：两种思路

很多人把这两个概念搞混。我简单解释一下：

• 滑动日志：记录每个请求的时间戳，每次判断时统计窗口内的请求数。精度高，但内存开销大。
• 滑动窗口：把时间窗口分成 N 个小格子，每个格子维护一个计数器。用格子粒度近似滑动效果。内存可控，精度取决于格子数。

对比维度	滑动日志	滑动窗口
精度	精确到毫秒	取决于格子数（格子越多越精确）
内存占用	O(请求数)，高并发下爆炸	O(格子数)，固定可控
实现复杂度	简单，但清理逻辑麻烦	中等，需要处理格子滑动
适用场景	低并发、高精度要求	高并发、生产环境主流方案

4.2 滑动窗口的核心逻辑

我画了一张图，帮你理解滑动窗口是怎么工作的。假设我们把 1 分钟切成 6 个格子，每个格子 10 秒。

每次新请求进来，我们做两件事：

1. 把当前时间戳对应的格子计数器 +1
2. 把窗口外的格子（比如第 0 个格子）计数器归零

然后统计所有格子的总和。如果超过阈值，就拒绝。

4.3 内存优化方案

滑动窗口虽然比滑动日志省内存，但也不是没有优化空间。我在一个日活千万的 API 网关项目里，把内存占用从 200MB 降到了 5MB。怎么做的？

方案一：环形数组

不用每次都创建新格子。用一个固定长度的数组，下标对格子数取模。每次滑动时，把当前下标对应的格子重置，而不是移动所有格子。

class SlidingWindowCounter:
    def __init__(self, window_size=60, bucket_count=10):
        self.window_size = window_size      # 窗口大小（秒）
        self.bucket_count = bucket_count    # 格子数
        self.bucket_span = window_size / bucket_count  # 每个格子跨度
        self.buckets = [0] * bucket_count   # 环形数组
        self.bucket_timestamps = [0] * bucket_count  # 每个格子的起始时间

方案二：懒清理

不主动清理过期格子。只在请求到来时，检查当前格子是否过期。如果过期，才重置。这样避免了定时器的开销。

def _get_current_bucket(self, now):
    index = int((now // self.bucket_span) % self.bucket_count)
    # 如果格子时间戳过期，重置
    if now - self.bucket_timestamps[index] >= self.bucket_span:
        self.buckets[index] = 0
        self.bucket_timestamps[index] = now
    return index

方案三：原子操作 + 无锁

高并发下，加锁是性能杀手。我建议用 Python 的 array 模块或者 ctypes 做原子自增。如果语言支持 CAS（Compare-And-Swap），直接用原子操作。

4.4 代码实战：Python 实现

下面是一个完整的滑动窗口限流器。我加了详细的注释，你可以直接拿去用。

import time
import threading

class SlidingWindowRateLimiter:
    """
    滑动窗口限流器
    支持：环形数组、懒清理、线程安全
    """
    def __init__(self, max_requests=100, window_size=60, bucket_count=10):
        self.max_requests = max_requests          # 窗口内最大请求数
        self.window_size = window_size            # 窗口大小（秒）
        self.bucket_count = bucket_count          # 格子数
        self.bucket_span = window_size / bucket_count  # 每个格子跨度

        # 环形数组：计数器和时间戳
        self.buckets = [0] * bucket_count
        self.bucket_timestamps = [0] * bucket_count

        # 锁（生产环境建议用读写锁）
        self.lock = threading.Lock()

    def _get_current_bucket_index(self, now):
        """计算当前时间对应的格子下标"""
        return int((now // self.bucket_span) % self.bucket_count)

    def _reset_bucket_if_expired(self, index, now):
        """如果格子过期，重置"""
        if now - self.bucket_timestamps[index] >= self.bucket_span:
            self.buckets[index] = 0
            self.bucket_timestamps[index] = now

    def allow_request(self) -> bool:
        """
        判断是否允许请求通过
        返回 True 表示放行，False 表示限流
        """
        now = time.time()
        with self.lock:
            # 1. 获取当前格子
            current_index = self._get_current_bucket_index(now)
            self._reset_bucket_if_expired(current_index, now)

            # 2. 统计窗口内所有格子的请求数
            total = 0
            for i in range(self.bucket_count):
                # 只统计未过期的格子
                if now - self.bucket_timestamps[i] < self.window_size:
                    total += self.buckets[i]

            # 3. 判断是否超限
            if total >= self.max_requests:
                return False

            # 4. 放行，计数器 +1
            self.buckets[current_index] += 1
            return True

    def get_current_count(self) -> int:
        """获取当前窗口内的请求数（用于监控）"""
        now = time.time()
        with self.lock:
            total = 0
            for i in range(self.bucket_count):
                if now - self.bucket_timestamps[i] < self.window_size:
                    total += self.buckets[i]
            return total

# 使用示例
if __name__ == "__main__":
    limiter = SlidingWindowRateLimiter(max_requests=10, window_size=60, bucket_count=6)

    # 模拟 15 个并发请求
    for i in range(15):
        result = limiter.allow_request()
        print(f"请求 {i+1}: {'✅ 放行' if result else '❌ 限流'}")
        time.sleep(0.1)

    print(f"\n当前窗口请求数: {limiter.get_current_count()}")

4.5 生产环境注意事项

• 时间同步：分布式环境下，各节点的时间戳必须同步。我建议用 NTP 服务，误差控制在 100ms 以内。
• 格子粒度：不要小于 100ms。格子太细，环形数组的取模运算会频繁触发，反而降低性能。
• 监控告警：一定要暴露 get_current_count() 接口。我在线上见过限流器默默工作，但没人知道阈值设得太低，导致大量正常请求被误杀。
• 降级策略：限流器本身不能成为单点故障。如果限流器挂了，应该默认放行（fail-open），而不是默认拒绝（fail-close）。

滑动窗口是限流算法里性价比最高的方案。它不像滑动日志那样吃内存，也不像固定窗口那样有毛刺。你只要把格子数、阈值、窗口大小调好，就能应对大部分场景。