4. 基础限流算法：固定窗口算法原理与Python实现

固定窗口算法（Fixed Window Algorithm）是限流领域最直观、最容易实现的算法之一。它将时间划分为固定的窗口（例如1秒、1分钟），在每个窗口内统计请求数量，一旦超过预设阈值则拒绝后续请求。虽然该算法存在临界突变问题，但因其实现简单、性能开销极低，在金融系统的非核心链路或对突发容忍度较高的场景中仍有广泛应用。

4.1 核心原理与数学模型

固定窗口算法的核心逻辑可以概括为三个要素：

• 窗口大小（Window Size）：时间窗口的长度，通常以毫秒或秒为单位。
• 阈值（Threshold）：单个窗口内允许通过的最大请求数。
• 计数器（Counter）：当前窗口内已累计的请求数。

算法流程如下：

1. 记录当前时间戳，计算其所属窗口的起始时间（current_time // window_size * window_size）。
2. 如果当前窗口起始时间与上次记录的窗口起始时间不同，则重置计数器为0。
3. 如果计数器小于阈值，则允许请求通过，计数器加1；否则拒绝请求。

4.2 固定窗口 vs 滑动窗口：核心差异

4.3 固定窗口算法的临界突变问题

固定窗口算法最著名的缺陷是窗口边界流量突变。假设窗口大小为1秒，阈值为10。如果在第0.9秒时累计了10个请求，第1.0秒窗口重置后，第1.1秒又可以处理10个请求。这意味着在0.2秒（0.9~1.1）的时间跨度内，系统可能承受20个请求，瞬时速率达到100 QPS，远超设定的10 QPS阈值。

4.4 固定窗口算法Python实现

以下实现采用线程安全设计，适用于多线程金融应用。使用 threading.Lock 保证计数器操作的原子性。

import time
import threading

class FixedWindowRateLimiter:
    """
    固定窗口限流器
    :param window_size: 窗口大小，单位秒
    :param threshold: 窗口内允许的最大请求数
    """
    def __init__(self, window_size: float = 1.0, threshold: int = 10):
        self.window_size = window_size
        self.threshold = threshold
        self._lock = threading.Lock()
        self._window_start = 0.0
        self._counter = 0

    def allow_request(self) -> bool:
        """
        判断当前请求是否允许通过
        返回 True 表示放行，False 表示拒绝
        """
        now = time.time()
        current_window = now // self.window_size * self.window_size

        with self._lock:
            # 窗口切换：重置计数器
            if current_window != self._window_start:
                self._window_start = current_window
                self._counter = 0

            # 判断是否超过阈值
            if self._counter < self.threshold:
                self._counter += 1
                return True
            else:
                return False

    def reset(self):
        """手动重置限流器状态"""
        with self._lock:
            self._window_start = 0.0
            self._counter = 0

4.5 使用示例与测试

以下代码演示了固定窗口限流器的基本使用，并模拟了窗口边界处的流量突变现象。

import time
import threading

def simulate_requests(limiter: FixedWindowRateLimiter, num_requests: int, interval: float):
    """模拟连续发送请求"""
    allowed = 0
    denied = 0
    for i in range(num_requests):
        if limiter.allow_request():
            allowed += 1
        else:
            denied += 1
        time.sleep(interval)
    return allowed, denied

# 创建限流器：窗口1秒，阈值5
limiter = FixedWindowRateLimiter(window_size=1.0, threshold=5)

# 测试1：正常窗口内限流
print("=== 测试1：窗口内连续请求 ===")
allowed, denied = simulate_requests(limiter, 10, 0.05)
print(f"允许: {allowed}, 拒绝: {denied}")  # 预期允许5，拒绝5

# 测试2：窗口边界突变演示
print("\n=== 测试2：窗口边界突变 ===")
limiter.reset()
# 在窗口末尾快速发送5个请求
for _ in range(5):
    limiter.allow_request()
    time.sleep(0.01)
# 等待窗口切换（剩余时间约0.95秒）
time.sleep(0.95)
# 在新窗口立即发送5个请求
allowed2, denied2 = simulate_requests(limiter, 5, 0.01)
print(f"新窗口允许: {allowed2}, 拒绝: {denied2}")  # 预期允许5，拒绝0
# 实际在约0.1秒内处理了10个请求，瞬时速率100 QPS

4.6 固定窗口算法架构示意

下图展示了固定窗口限流器在金融系统网关层的位置，以及请求在窗口内的处理流程。

4.7 固定窗口算法在金融系统中的适用性评估