3、计数器算法详解：固定窗口计数器实现、临界问题分析、代码实战（Python实现）

限流算法里，计数器算法是最直观的一种。说白了，就是在一个时间窗口内数请求次数，超过了就拦。我刚开始做API网关时，第一个限流方案用的就是它——简单、好理解、容易实现。

但你别因为它简单就小看它。用不好，坑也不少。今天咱们就把固定窗口计数器掰开揉碎了讲清楚。

3.1 固定窗口计数器原理

固定窗口计数器的思路特别直白：

• 把时间切成固定大小的窗口（比如1秒、1分钟）
• 每个窗口维护一个计数器
• 请求来了，计数器加1
• 计数器超过阈值，直接拒绝
• 窗口结束，计数器归零

举个例子：我限制某个API每秒最多100次请求。那么从00:00:00到00:00:01这1秒内，计数器从0开始，每来一个请求加1。到100了，后面的请求全部返回429。等到了00:00:01，计数器重置，重新计数。

3.2 临界问题（Traffic Burst at Boundaries）

固定窗口有个著名的坑——临界问题。我当年第一次上线就踩了这个雷。

为什么会这样？你想想看：

假设我限制QPS=100。窗口A（00:00:00-00:00:01）在最后100ms内来了100个请求，窗口B（00:00:01-00:00:02）在最开始100ms内又来了100个请求。从单个窗口看，都没超限。但从连续时间看，这200ms内实际处理了200个请求——QPS瞬间飙到1000。

这就是典型的窗口边界突发流量。

时间点	窗口A计数器	窗口B计数器	实际QPS
00:00:00.900	99	0	—
00:00:00.950	100（满）	0	—
00:00:01.050	重置为0	100（满）	200/0.2s = 1000

解决思路有两个方向：

• 滑动窗口：把窗口粒度切得更细，比如1秒切成10个100ms的小格子，用滑动的方式统计——这个后面章节会细讲
• 加随机延迟：在窗口边界附近引入少量随机等待，打散突发——但治标不治本

3.3 代码实战：Python实现固定窗口计数器

下面我写一个生产可用的固定窗口计数器。注意，我加了线程安全处理，因为API限流通常是多线程环境。

import time
import threading

class FixedWindowRateLimiter:
    """
    固定窗口计数器限流器
    支持多线程安全
    """
    def __init__(self, max_requests: int, window_size: float = 1.0):
        """
        :param max_requests: 窗口内允许的最大请求数
        :param window_size: 窗口大小，单位秒，默认1秒
        """
        self.max_requests = max_requests
        self.window_size = window_size
        self.lock = threading.Lock()
        
        # 当前窗口信息
        self.current_window_start = time.time()
        self.counter = 0
    
    def allow_request(self) -> bool:
        """
        判断当前请求是否允许通过
        返回True表示允许，False表示拒绝
        """
        now = time.time()
        
        with self.lock:
            # 检查是否进入新的时间窗口
            if now - self.current_window_start >= self.window_size:
                # 重置窗口
                self.current_window_start = now
                self.counter = 0
            
            # 判断是否超限
            if self.counter < self.max_requests:
                self.counter += 1
                return True
            else:
                return False
    
    def get_current_count(self) -> int:
        """获取当前窗口的请求计数"""
        with self.lock:
            return self.counter

# ========== 使用示例 ==========
if __name__ == "__main__":
    # 限制每秒最多5个请求
    limiter = FixedWindowRateLimiter(max_requests=5, window_size=1.0)
    
    # 模拟并发请求
    def send_request(request_id):
        if limiter.allow_request():
            print(f"[{time.strftime('%H:%M:%S')}] 请求 {request_id}: ✅ 通过")
        else:
            print(f"[{time.strftime('%H:%M:%S')}] 请求 {request_id}: ❌ 被限流")
    
    # 启动10个线程同时发请求
    threads = []
    for i in range(10):
        t = threading.Thread(target=send_request, args=(i,))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    print(f"\n当前窗口计数: {limiter.get_current_count()}")

3.4 固定窗口 vs 其他算法：一张图看懂

下面这张SVG图把固定窗口的核心逻辑和临界问题画清楚了。

3.5 适用场景与局限性

固定窗口计数器不是万能的，但它有自己的一亩三分地。

维度	说明
✅ 适用场景	对突发流量不敏感的业务；日配额/小时配额等粗粒度限流；内部系统间的调用限制
❌ 不适用场景	对延迟敏感的实时接口；需要精确控制瞬时QPS的场景；秒杀、抢购等流量尖峰场景
💡 我的建议	如果业务能接受2倍左右的突发，固定窗口完全够用。别为了追求完美而过度设计。