4. 漏桶算法原理与实现

漏桶算法（Leaky Bucket）是流量整形与限流领域的经典算法，其核心思想是将突发流量“削峰填谷”，使输出流量以恒定速率流出，从而保护下游系统（如模型推理服务、数据库）免受流量冲击。在模型风险管理场景中，漏桶算法常用于控制API调用频率，确保模型服务在SLA内稳定响应。

4.1 漏桶算法的工作机制

漏桶算法模拟一个底部有漏洞的桶：水（请求）可以以任意速率注入桶中，但桶底漏洞以固定速率漏水（处理请求）。当桶满时，新注入的水会溢出（请求被拒绝）。

工作流程如下：

1. 每个请求到达时，首先根据当前时间与上次漏水时间的时间差，计算应漏掉的水量，并更新当前水量。
2. 判断当前水量是否小于桶容量：若是，则请求被接受，水量加1；若否，则请求被拒绝（溢出）。
3. 系统以恒定速率从桶中取出请求进行处理，不受注入速率影响。

4.2 漏桶算法与令牌桶算法的对比

漏桶算法与令牌桶算法是限流领域最常用的两种算法，二者在流量整形逻辑上存在本质差异。下表从多个维度进行对比：

4.3 Python代码实现

以下实现一个线程安全的漏桶算法，适用于模型风险管理API的限流场景。代码包含核心类、请求处理逻辑和单元测试示例。

import time
import threading

class LeakyBucket:
    """漏桶算法实现（线程安全）"""
    
    def __init__(self, capacity: int, leak_rate: float):
        """
        :param capacity: 桶容量，最大允许积压请求数
        :param leak_rate: 漏水速率，单位：请求/秒
        """
        self.capacity = capacity
        self.leak_rate = leak_rate
        self._water = 0.0          # 当前水量
        self._last_leak_time = time.monotonic()  # 上次漏水时间戳
        self._lock = threading.Lock()
    
    def _leak(self):
        """根据时间差计算应漏掉的水量"""
        now = time.monotonic()
        elapsed = now - self._last_leak_time
        leaked = elapsed * self.leak_rate
        self._water = max(0.0, self._water - leaked)
        self._last_leak_time = now
    
    def allow_request(self) -> bool:
        """
        判断是否允许请求通过
        :return: True 表示允许，False 表示拒绝
        """
        with self._lock:
            self._leak()
            if self._water < self.capacity:
                self._water += 1.0
                return True
            else:
                return False
    
    def get_water_level(self) -> float:
        """获取当前水量（调试用）"""
        with self._lock:
            self._leak()
            return self._water

# ========== 使用示例 ==========
if __name__ == "__main__":
    # 创建漏桶：容量10，每秒处理5个请求
    bucket = LeakyBucket(capacity=10, leak_rate=5.0)
    
    # 模拟突发请求
    for i in range(20):
        allowed = bucket.allow_request()
        status = "✅ 允许" if allowed else "❌ 拒绝"
        print(f"请求 {i+1:2d}: {status} | 当前水量: {bucket.get_water_level():.1f}")
        time.sleep(0.05)  # 模拟请求间隔50ms
    
    # 等待漏水，观察水量下降
    print("\n等待1秒后...")
    time.sleep(1.0)
    print(f"当前水量: {bucket.get_water_level():.1f}")

4.4 模型风险管理中的实践要点