4、漏桶算法详解

漏桶算法（Leaky Bucket）是流量整形与限流领域的经典算法。其核心思想是将突发流量“削峰填谷”，强制数据以恒定速率流出，从而保护下游系统（如数据库、支付网关）免受流量冲击。在金融API场景中，漏桶算法常用于控制对账请求、交易流水写入等需要严格平滑速率的操作。

4.1 工作机制

漏桶算法模拟一个底部有漏洞的桶：

• 桶容量（Capacity）：桶的最大容量，代表允许的最大突发请求数。
• 漏水速率（Leak Rate）：单位时间内从桶底漏出的请求数，即系统处理请求的恒定速率。
• 流入请求：请求到达时，如果桶未满，则加入桶中等待处理；如果桶已满，则请求被丢弃（拒绝服务）。
• 恒定流出：无论流入速率如何波动，系统始终以漏水速率从桶中取出请求进行处理。

4.2 与令牌桶算法的区别

漏桶算法与令牌桶算法是限流领域最常对比的两种算法。下表从多个维度进行对比：

4.3 Python代码实现

以下是一个基于时间戳的漏桶算法实现，无需依赖外部队列库，适合嵌入金融API中间件：

import time
import threading

class LeakyBucket:
    """
    漏桶算法实现
    capacity: 桶容量（最大排队请求数）
    leak_rate: 漏水速率（每秒处理请求数）
    """
    def __init__(self, capacity: int, leak_rate: float):
        self.capacity = capacity
        self.leak_rate = leak_rate
        self._water = 0          # 当前桶中水量
        self._last_leak_time = time.time()
        self._lock = threading.Lock()

    def _leak(self):
        """执行漏水操作，根据时间差计算漏出水量"""
        now = time.time()
        elapsed = now - self._last_leak_time
        # 计算应漏出的水量
        leaked = elapsed * self.leak_rate
        self._water = max(0, self._water - leaked)
        self._last_leak_time = now

    def allow_request(self) -> bool:
        """
        判断是否允许请求通过
        返回 True 表示允许，False 表示拒绝
        """
        with self._lock:
            self._leak()
            if self._water < self.capacity:
                self._water += 1
                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)

    # 模拟20个并发请求
    for i in range(20):
        if bucket.allow_request():
            print(f"请求 {i+1}: 通过")
        else:
            print(f"请求 {i+1}: 被限流（丢弃）")
        time.sleep(0.05)  # 模拟请求间隔

4.4 适用场景

漏桶算法在金融科技领域有明确的适用边界：

4.5 金融API测试中的验证要点

在测试漏桶算法实现的限流组件时，应重点验证以下指标：

1. 恒定输出速率验证：在长时间（如60秒）高并发压力下，测量后端实际接收到的请求速率，确认其稳定在 leak_rate 附近，波动不超过 ±5%。
2. 桶容量边界测试：发送超过桶容量的突发请求，验证被拒绝的请求数是否等于 请求总数 - 桶容量。
3. 空闲后恢复测试：停止发送请求一段时间（如10秒），然后重新发送，验证桶能重新接受请求（水量已漏空）。
4. 并发安全测试：使用多线程（如50线程）同时调用 allow_request()，验证无竞态条件导致的水量计算错误。