4、漏桶算法：恒定速率处理请求的机制、Python代码实现与适用场景

漏桶算法（Leaky Bucket Algorithm）是金融API限流中最直观的流量整形算法之一。它将突发流量比作注入桶中的水，桶底以恒定速率“漏水”，从而将不规则的请求流平滑为固定速率的输出流。在支付、风控、行情推送等对请求间隔有严格要求的场景中，漏桶算法能有效防止后端系统被瞬时流量冲垮。

4.1 核心机制与数学模型

漏桶算法的核心思想包含两个关键参数：

• 桶容量（Capacity）：桶能容纳的最大请求数，对应系统能承受的瞬时突发上限。
• 漏水速率（Leak Rate）：单位时间内从桶底漏出的请求数，即系统处理请求的恒定速率。

当新请求到达时，算法执行以下逻辑：

1. 检查当前桶中的水量（积压请求数）。
2. 如果桶已满（水量 ≥ 容量），则拒绝该请求（丢弃或返回限流错误）。
3. 如果桶未满，将请求放入桶中，等待以恒定速率被“漏出”处理。

4.2 漏桶 vs 令牌桶：核心差异

在限流算法家族中，漏桶算法常与令牌桶算法对比。下表从多个维度厘清两者的本质区别：

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_time = time.monotonic()  # 上次漏水时间戳
        self._lock = threading.Lock()
    
    def _leak(self):
        """根据时间差计算漏掉的水量"""
        now = time.monotonic()
        elapsed = now - self._last_time
        leaked = elapsed * self.leak_rate
        self._water = max(0.0, self._water - leaked)
        self._last_time = now
    
    def allow_request(self) -> bool:
        """
        判断是否允许当前请求通过
        :return: True 允许，False 拒绝
        """
        with self._lock:
            self._leak()
            if self._water + 1 <= self.capacity:
                self._water += 1
                return True
            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)
    for i in range(5):
        allowed = bucket.allow_request()
        status = "✅ 通过" if allowed else "❌ 拒绝"
        print(f"请求 {i+1:2d}: {status} | 当前水量: {bucket.get_water_level():.1f}")
        time.sleep(0.1)

4.4 适用场景与避坑指南

漏桶算法在金融系统中主要应用于以下场景：

4.5 性能与监控指标

在生产环境中部署漏桶限流器后，建议持续监控以下指标：

• 当前水量（water level）：反映系统积压压力，接近容量时触发告警。
• 拒绝率（reject rate）：被漏桶拒绝的请求占比，过高说明容量或速率需要调整。
• 实际处理速率（actual throughput）：对比漏水速率，验证限流效果。
• 漏水延迟（leak latency）：从请求入桶到被处理的时间，用于评估排队等待时长。