2、API限流基础概念：限流的目的、限流与熔断的区别、常见限流算法

在模型风险管理系统中，API 是模型服务对外暴露的唯一窗口。当突发流量、恶意攻击或上游系统故障导致请求量激增时，若不加控制，模型服务可能因资源耗尽而崩溃，甚至引发级联故障。限流（Rate Limiting）与熔断（Circuit Breaking）是保障系统稳定性的两大核心手段，但二者在目标、触发机制和恢复方式上存在本质区别。

2.1 限流的目的

限流的核心目标是主动控制请求速率，确保系统在安全阈值内平稳运行。具体目的包括：

• 保护后端资源：防止模型推理服务因 CPU、内存、GPU 或数据库连接池耗尽而崩溃。
• 保障公平使用：避免某个调用方（如高频交易策略）独占资源，导致其他业务方被饿死。
• 抵御突发流量：平滑尖峰请求，防止瞬时流量冲垮系统。
• 成本控制：在云原生环境中，限流可避免因无限制调用导致的计算资源费用失控。
• 满足 SLA：通过限制并发数或 QPS，保证核心请求的响应时间在约定范围内。

2.2 限流与熔断的区别

限流和熔断经常被混淆，但它们在故障处理链路中扮演不同角色。下表从多个维度进行对比：

2.3 常见限流算法

不同的限流算法在精度、内存开销和应对突发流量的能力上各有优劣。以下三种算法是业界最常用的方案。

2.3.1 令牌桶算法（Token Bucket）

令牌桶算法以固定速率向桶中放入令牌，每个请求需获取一个令牌才能被处理。桶有最大容量，允许一定程度的突发流量。其核心参数包括：

• 令牌生成速率（r）：每秒向桶中添加的令牌数，决定长期平均速率。
• 桶容量（b）：桶中最多存放的令牌数，决定突发流量的大小。
• 当前令牌数（c）：实时可用令牌数量。

算法流程：

1. 系统以固定速率 r 向桶中添加令牌，桶满则丢弃多余令牌。
2. 请求到达时，检查桶中是否有可用令牌（c > 0）。
3. 若有，则消耗一个令牌，请求通过；若无，则请求被拒绝或排队。

2.3.2 漏桶算法（Leaky Bucket）

漏桶算法将请求视为水滴，以固定速率从桶底漏出处理，桶满则溢出（拒绝）。与令牌桶不同，漏桶强制平滑流量，不允许突发。其核心参数包括：

• 漏出速率（r）：每秒从桶底流出的请求数，即处理速率。
• 桶容量（b）：桶中最多容纳的等待请求数。
• 当前水量（w）：桶中正在等待的请求数。

2.3.3 计数器算法（Counter / Fixed Window）

计数器算法是最直观的限流方式：将时间划分为固定窗口（如 1 秒），统计窗口内的请求数，超过阈值则拒绝。其核心参数为：

• 窗口大小（w）：时间窗口长度，通常为 1 秒或 1 分钟。
• 窗口阈值（t）：每个窗口内允许的最大请求数。
• 当前计数（c）：当前窗口已累计的请求数。

2.3.4 三种算法对比

2.3.5 代码示例：令牌桶算法核心逻辑

以下为令牌桶算法的简化实现，展示了核心的令牌消耗与补充逻辑：

import time

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate                # 令牌生成速率（个/秒）
        self.capacity = capacity        # 桶容量
        self.tokens = capacity          # 当前令牌数
        self.last_refill = time.time()  # 上次补充时间

    def _refill(self):
        """按时间间隔补充令牌"""
        now = time.time()
        elapsed = now - self.last_refill
        new_tokens = elapsed * self.rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now

    def try_acquire(self, count: int = 1) -> bool:
        """尝试获取 count 个令牌，成功返回 True"""
        self._refill()
        if self.tokens >= count:
            self.tokens -= count
            return True
        return False

# 使用示例：每秒 100 个令牌，桶容量 200
bucket = TokenBucket(rate=100, capacity=200)
for i in range(300):
    if bucket.try_acquire():
        print(f"请求 {i}: 通过")
    else:
        print(f"请求 {i}: 限流")
    time.sleep(0.005)  # 模拟 5ms 间隔