2. 限流算法基础：计数器算法原理、Python实现与临界问题分析

计数器算法是限流领域最直观、最容易实现的算法。它通过维护一个时间窗口内的请求计数器，当计数器超过预设阈值时拒绝后续请求，直到窗口重置。本节将深入剖析其原理、Python实现方式以及经典的临界问题。

2.1 计数器算法核心原理

计数器算法的核心思想可以概括为：固定时间窗口 + 原子计数器。其工作流程如下：

1. 定义时间窗口：设定一个固定的时间长度，例如1秒、1分钟或1小时。
2. 初始化计数器：在每个窗口开始时，将计数器重置为0。
3. 请求计数：每收到一个请求，计数器加1。
4. 阈值判断：如果计数器值小于等于设定的阈值（如100次/秒），则允许请求通过；否则，拒绝请求。
5. 窗口重置：当前时间窗口结束后，计数器归零，进入下一个窗口。

2.2 固定窗口计数器 Python 实现

以下是一个基于 time 模块的固定窗口计数器实现，使用 threading.Lock 保证线程安全：

import time
import threading

class FixedWindowCounter:
    """固定窗口计数器限流器"""

    def __init__(self, limit: int, window_size: float = 1.0):
        """
        :param limit: 窗口内允许的最大请求数
        :param window_size: 时间窗口大小（秒），默认1秒
        """
        self.limit = limit
        self.window_size = window_size
        self.counter = 0
        self.window_start = time.time()
        self.lock = threading.Lock()

    def allow_request(self) -> bool:
        """判断当前请求是否允许通过"""
        with self.lock:
            now = time.time()
            # 如果当前时间已超出窗口，重置计数器
            if now - self.window_start >= self.window_size:
                self.counter = 0
                self.window_start = now

            if self.counter < self.limit:
                self.counter += 1
                return True
            else:
                return False

# 使用示例：限制每秒最多5个请求
limiter = FixedWindowCounter(limit=5, window_size=1.0)

for i in range(10):
    if limiter.allow_request():
        print(f"请求 {i+1}: 通过")
    else:
        print(f"请求 {i+1}: 被限流")
    time.sleep(0.1)  # 模拟请求间隔

2.3 临界问题（Traffic Burst at Window Boundary）

固定窗口计数器存在一个显著的缺陷——临界问题。由于窗口切换瞬间计数器归零，可能导致在窗口边界处出现流量尖峰，使实际通过的请求数在短时间内远超阈值。

假设限流策略为 1秒内最多允许100个请求。攻击者可以在第0.9秒到第1.0秒之间发送100个请求（窗口A末尾），紧接着在第1.0秒到第1.1秒之间再发送100个请求（窗口B开头）。在0.2秒的时间跨度内，后端实际接收了200个请求，完全绕过了限流策略。

2.4 滑动窗口计数器：临界问题的改进方案

为了解决固定窗口的临界问题，业界提出了滑动窗口计数器算法。它将时间窗口细分为多个更小的子窗口（例如将1秒分为10个100ms的子窗口），通过维护子窗口的计数器来更平滑地计算请求总量。

滑动窗口的核心改进点：

• 细粒度统计：不再以整个窗口为单位重置计数器，而是滚动计算最近一个完整窗口内的请求总数。
• 平滑限流：窗口边界处的流量突变被多个子窗口分摊，不会出现瞬间计数器归零导致的流量尖峰。
• 精度可调：子窗口数量越多，限流精度越高，但内存开销也相应增加。

2.5 计数器算法对比总结